More work on the pds notifs

src/api/admin/mod.rs

@@ -1,5 +1,6 @@
 pub mod account;
 pub mod invite;
+pub mod server_stats;
 pub mod status;
 
 pub use account::{
@@ -9,4 +10,5 @@ pub use account::{
 pub use invite::{
     disable_account_invites, disable_invite_codes, enable_account_invites, get_invite_codes,
 };
+pub use server_stats::get_server_stats;
 pub use status::{get_subject_status, update_subject_status};

src/api/admin/server_stats.rs

@@ -0,0 +1,76 @@
+use crate::state::AppState;
+use axum::{
+    Json,
+    extract::State,
+    http::{HeaderMap, StatusCode},
+    response::{IntoResponse, Response},
+};
+use serde::Serialize;
+use serde_json::json;
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ServerStatsResponse {
+    pub user_count: i64,
+    pub repo_count: i64,
+    pub record_count: i64,
+    pub blob_storage_bytes: i64,
+}
+
+pub async fn get_server_stats(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> Response {
+    let auth_header = headers.get("Authorization");
+    if auth_header.is_none() {
+        return (
+            StatusCode::UNAUTHORIZED,
+            Json(json!({"error": "AuthenticationRequired"})),
+        )
+            .into_response();
+    }
+
+    let user_count: i64 = match sqlx::query_scalar!("SELECT COUNT(*) FROM users")
+        .fetch_one(&state.db)
+        .await
+    {
+        Ok(Some(count)) => count,
+        Ok(None) => 0,
+        Err(_) => 0,
+    };
+
+    let repo_count: i64 = match sqlx::query_scalar!("SELECT COUNT(*) FROM repos")
+        .fetch_one(&state.db)
+        .await
+    {
+        Ok(Some(count)) => count,
+        Ok(None) => 0,
+        Err(_) => 0,
+    };
+
+    let record_count: i64 = match sqlx::query_scalar!("SELECT COUNT(*) FROM records")
+        .fetch_one(&state.db)
+        .await
+    {
+        Ok(Some(count)) => count,
+        Ok(None) => 0,
+        Err(_) => 0,
+    };
+
+    let blob_storage_bytes: i64 = match sqlx::query_scalar!("SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM blobs")
+        .fetch_one(&state.db)
+        .await
+    {
+        Ok(Some(bytes)) => bytes,
+        Ok(None) => 0,
+        Err(_) => 0,
+    };
+
+    Json(ServerStatsResponse {
+        user_count,
+        repo_count,
+        record_count,
+        blob_storage_bytes,
+    })
+    .into_response()
+}

src/api/identity/account.rs

@@ -382,17 +382,14 @@ pub async fn create_account(
     let user_insert: Result<(uuid::Uuid,), _> = sqlx::query_as(
         r#"INSERT INTO users (
             handle, email, did, password_hash,
-            email_confirmation_code, email_confirmation_code_expires_at,
             preferred_notification_channel,
             discord_id, telegram_username, signal_number
-        ) VALUES ($1, $2, $3, $4, $5, $6, $7::notification_channel, $8, $9, $10) RETURNING id"#,
+        ) VALUES ($1, $2, $3, $4, $5::notification_channel, $6, $7, $8) RETURNING id"#,
     )
     .bind(short_handle)
     .bind(&email)
     .bind(&did)
     .bind(&password_hash)
-    .bind(&verification_code)
-    .bind(code_expires_at)
     .bind(verification_channel)
     .bind(
         input
@@ -460,6 +457,23 @@ pub async fn create_account(
                 .into_response();
         }
     };
+
+    if let Err(e) = sqlx::query!(
+        "INSERT INTO channel_verifications (user_id, channel, code, pending_identifier, expires_at) VALUES ($1, 'email', $2, $3, $4)",
+        user_id,
+        verification_code,
+        email,
+        code_expires_at
+    )
+    .execute(&mut *tx)
+    .await {
+        error!("Error inserting verification code: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response();
+    }
     let encrypted_key_bytes = match crate::config::encrypt_key(&secret_key_bytes) {
         Ok(enc) => enc,
         Err(e) => {

src/api/mod.rs

@@ -13,6 +13,7 @@ pub mod repo;
 pub mod server;
 pub mod temp;
 pub mod validation;
+pub mod verification;
 
 pub use error::ApiError;
 pub use proxy_client::{AtUriParts, proxy_client, validate_at_uri, validate_did, validate_limit};

src/api/notification_prefs.rs

@@ -6,11 +6,21 @@ use axum::{
     http::{HeaderMap, StatusCode},
     response::{IntoResponse, Response},
 };
+use chrono::{Duration, Utc};
+use rand::Rng;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use sqlx::Row;
 use tracing::info;
 
+fn generate_verification_code() -> String {
+    rand::thread_rng()
+        .sample_iter(&rand::distributions::Uniform::new(0, 10))
+        .take(6)
+        .map(|x| x.to_string())
+        .collect()
+}
+
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct NotificationPrefsResponse {
@@ -95,15 +105,172 @@ pub async fn get_notification_prefs(State(state): State<AppState>, headers: Head
     .into_response()
 }
 
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct NotificationHistoryEntry {
+    pub created_at: String,
+    pub channel: String,
+    pub notification_type: String,
+    pub status: String,
+    pub subject: Option<String>,
+    pub body: String,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetNotificationHistoryResponse {
+    pub notifications: Vec<NotificationHistoryEntry>,
+}
+
+pub async fn get_notification_history(
+    State(state): State<AppState>,
+    headers: 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", "message": "Authentication required"})),
+        )
+            .into_response(),
+    };
+    let user = match validate_bearer_token(&state.db, &token).await {
+        Ok(u) => u,
+        Err(_) => {
+            return (
+                StatusCode::UNAUTHORIZED,
+                Json(json!({"error": "AuthenticationFailed", "message": "Invalid token"})),
+            )
+                .into_response();
+        }
+    };
+
+    let user_id: uuid::Uuid = match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", user.did)
+        .fetch_one(&state.db)
+        .await
+    {
+        Ok(id) => id,
+        Err(e) => return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        )
+            .into_response(),
+    };
+
+    let rows = match sqlx::query!(
+        r#"
+        SELECT
+            created_at,
+            channel as "channel: String",
+            notification_type as "notification_type: String",
+            status as "status: String",
+            subject,
+            body
+        FROM notification_queue
+        WHERE user_id = $1
+        ORDER BY created_at DESC
+        LIMIT 50
+        "#,
+        user_id
+    )
+    .fetch_all(&state.db)
+    .await
+    {
+        Ok(r) => r,
+        Err(e) => return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        )
+            .into_response(),
+    };
+
+    let notifications = rows.iter().map(|row| {
+        NotificationHistoryEntry {
+            created_at: row.created_at.to_rfc3339(),
+            channel: row.channel.clone(),
+            notification_type: row.notification_type.clone(),
+            status: row.status.clone(),
+            subject: row.subject.clone(),
+            body: row.body.clone(),
+        }
+    }).collect();
+
+    Json(GetNotificationHistoryResponse { notifications }).into_response()
+}
+
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct UpdateNotificationPrefsInput {
     pub preferred_channel: Option<String>,
+    pub email: Option<String>,
     pub discord_id: Option<String>,
     pub telegram_username: Option<String>,
     pub signal_number: Option<String>,
 }
 
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct UpdateNotificationPrefsResponse {
+    pub success: bool,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub verification_required: Vec<String>,
+}
+
+pub async fn request_channel_verification(
+    db: &sqlx::PgPool,
+    user_id: uuid::Uuid,
+    channel: &str,
+    identifier: &str,
+    handle: Option<&str>,
+) -> Result<String, String> {
+    let code = generate_verification_code();
+    let expires_at = Utc::now() + Duration::minutes(10);
+
+    sqlx::query!(
+        r#"
+        INSERT INTO channel_verifications (user_id, channel, code, pending_identifier, expires_at)
+        VALUES ($1, $2::notification_channel, $3, $4, $5)
+        ON CONFLICT (user_id, channel) DO UPDATE
+        SET code = $3, pending_identifier = $4, expires_at = $5, created_at = NOW()
+        "#,
+        user_id,
+        channel as _,
+        code,
+        identifier,
+        expires_at
+    )
+    .execute(db)
+    .await
+    .map_err(|e| format!("Database error: {}", e))?;
+
+    if channel == "email" {
+        let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
+        let handle_str = handle.unwrap_or("user");
+        crate::notifications::enqueue_email_update(db, user_id, identifier, handle_str, &code, &hostname)
+            .await
+            .map_err(|e| format!("Failed to enqueue email notification: {}", e))?;
+    } else {
+        sqlx::query!(
+            r#"
+            INSERT INTO notification_queue (user_id, channel, notification_type, recipient, subject, body, metadata)
+            VALUES ($1, $2::notification_channel, 'channel_verification', $3, 'Verify your channel', $4, $5)
+            "#,
+            user_id,
+            channel as _,
+            identifier,
+            format!("Your verification code is: {}", code),
+            json!({"code": code})
+        )
+        .execute(db)
+        .await
+        .map_err(|e| format!("Failed to enqueue notification: {}", e))?;
+    }
+
+    Ok(code)
+}
+
 pub async fn update_notification_prefs(
     State(state): State<AppState>,
     headers: HeaderMap,
@@ -129,6 +296,28 @@ pub async fn update_notification_prefs(
                 .into_response();
         }
     };
+
+    let user_row = match sqlx::query!(
+        "SELECT id, handle, email FROM users WHERE did = $1",
+        user.did
+    )
+    .fetch_one(&state.db)
+    .await
+    {
+        Ok(row) => row,
+        Err(e) => return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        )
+            .into_response(),
+    };
+
+    let user_id = user_row.id;
+    let handle = user_row.handle;
+    let current_email = user_row.email;
+
+    let mut verification_required: Vec<String> = Vec::new();
+
     if let Some(ref channel) = input.preferred_channel {
         let valid_channels = ["email", "discord", "telegram", "signal"];
         if !valid_channels.contains(&channel.as_str()) {
@@ -157,71 +346,164 @@ pub async fn update_notification_prefs(
         }
         info!(did = %user.did, channel = %channel, "Updated preferred notification channel");
     }
+
+    if let Some(ref new_email) = input.email {
+        let email_clean = new_email.trim().to_lowercase();
+        if email_clean.is_empty() {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidRequest", "message": "Email cannot be empty"})),
+            )
+                .into_response();
+        }
+
+        if !crate::api::validation::is_valid_email(&email_clean) {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidEmail", "message": "Invalid email format"})),
+            )
+                .into_response();
+        }
+
+        if current_email.as_ref().map(|e| e.to_lowercase()) == Some(email_clean.clone()) {
+            info!(did = %user.did, "Email unchanged, skipping");
+        } else {
+            let exists = sqlx::query!(
+                "SELECT 1 as one FROM users WHERE LOWER(email) = $1 AND id != $2",
+                email_clean,
+                user_id
+            )
+            .fetch_optional(&state.db)
+            .await;
+
+            if let Ok(Some(_)) = exists {
+                return (
+                    StatusCode::BAD_REQUEST,
+                    Json(json!({"error": "EmailTaken", "message": "Email already in use"})),
+                )
+                    .into_response();
+            }
+
+            if let Err(e) = request_channel_verification(&state.db, user_id, "email", &email_clean, Some(&handle)).await {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": e})),
+                )
+                    .into_response();
+            }
+            verification_required.push("email".to_string());
+            info!(did = %user.did, "Requested email verification");
+        }
+    }
+
     if let Some(ref discord_id) = input.discord_id {
-        let discord_id_clean: Option<&str> = if discord_id.is_empty() {
-            None
-        } else {
-            Some(discord_id.as_str())
-        };
-        if let Err(e) = sqlx::query(
-            r#"UPDATE users SET discord_id = $1, discord_verified = FALSE, updated_at = NOW() WHERE did = $2"#
-        )
-        .bind(discord_id_clean)
-        .bind(&user.did)
-        .execute(&state.db)
-        .await
-        {
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        if discord_id.is_empty() {
+            if let Err(e) = sqlx::query!(
+                "UPDATE users SET discord_id = NULL, discord_verified = FALSE, updated_at = NOW() WHERE id = $1",
+                user_id
             )
-                .into_response();
+            .execute(&state.db)
+            .await
+            {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+                )
+                    .into_response();
+            }
+            let _ = sqlx::query!(
+                "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = 'discord'",
+                user_id
+            )
+            .execute(&state.db)
+            .await;
+            info!(did = %user.did, "Cleared Discord ID");
+        } else {
+            if let Err(e) = request_channel_verification(&state.db, user_id, "discord", discord_id, None).await {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": e})),
+                )
+                    .into_response();
+            }
+            verification_required.push("discord".to_string());
+            info!(did = %user.did, "Requested Discord verification");
         }
-        info!(did = %user.did, "Updated Discord ID");
     }
+
     if let Some(ref telegram) = input.telegram_username {
-        let telegram_clean: Option<&str> = if telegram.is_empty() {
-            None
-        } else {
-            Some(telegram.trim_start_matches('@'))
-        };
-        if let Err(e) = sqlx::query(
-            r#"UPDATE users SET telegram_username = $1, telegram_verified = FALSE, updated_at = NOW() WHERE did = $2"#
-        )
-        .bind(telegram_clean)
-        .bind(&user.did)
-        .execute(&state.db)
-        .await
-        {
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        let telegram_clean = telegram.trim_start_matches('@');
+        if telegram_clean.is_empty() {
+            if let Err(e) = sqlx::query!(
+                "UPDATE users SET telegram_username = NULL, telegram_verified = FALSE, updated_at = NOW() WHERE id = $1",
+                user_id
             )
-                .into_response();
+            .execute(&state.db)
+            .await
+            {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+                )
+                    .into_response();
+            }
+            let _ = sqlx::query!(
+                "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = 'telegram'",
+                user_id
+            )
+            .execute(&state.db)
+            .await;
+            info!(did = %user.did, "Cleared Telegram username");
+        } else {
+            if let Err(e) = request_channel_verification(&state.db, user_id, "telegram", telegram_clean, None).await {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": e})),
+                )
+                    .into_response();
+            }
+            verification_required.push("telegram".to_string());
+            info!(did = %user.did, "Requested Telegram verification");
         }
-        info!(did = %user.did, "Updated Telegram username");
     }
+
     if let Some(ref signal) = input.signal_number {
-        let signal_clean: Option<&str> = if signal.is_empty() {
-            None
-        } else {
-            Some(signal.as_str())
-        };
-        if let Err(e) = sqlx::query(
-            r#"UPDATE users SET signal_number = $1, signal_verified = FALSE, updated_at = NOW() WHERE did = $2"#
-        )
-        .bind(signal_clean)
-        .bind(&user.did)
-        .execute(&state.db)
-        .await
-        {
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        if signal.is_empty() {
+            if let Err(e) = sqlx::query!(
+                "UPDATE users SET signal_number = NULL, signal_verified = FALSE, updated_at = NOW() WHERE id = $1",
+                user_id
             )
-                .into_response();
+            .execute(&state.db)
+            .await
+            {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+                )
+                    .into_response();
+            }
+            let _ = sqlx::query!(
+                "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = 'signal'",
+                user_id
+            )
+            .execute(&state.db)
+            .await;
+            info!(did = %user.did, "Cleared Signal number");
+        } else {
+            if let Err(e) = request_channel_verification(&state.db, user_id, "signal", signal, None).await {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": e})),
+                )
+                    .into_response();
+            }
+            verification_required.push("signal".to_string());
+            info!(did = %user.did, "Requested Signal verification");
         }
-        info!(did = %user.did, "Updated Signal number");
     }
-    Json(json!({"success": true})).into_response()
+
+    Json(UpdateNotificationPrefsResponse {
+        success: true,
+        verification_required,
+    }).into_response()
 }

src/api/server/email.rs

@@ -6,15 +6,11 @@ use axum::{
     http::StatusCode,
     response::{IntoResponse, Response},
 };
-use chrono::{Duration, Utc};
+use chrono::Utc;
 use serde::Deserialize;
 use serde_json::json;
 use tracing::{error, info, warn};
 
-fn generate_confirmation_code() -> String {
-    crate::util::generate_token_code()
-}
-
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct RequestEmailUpdateInput {
@@ -41,6 +37,7 @@ pub async fn request_email_update(
         )
             .into_response();
     }
+
     let token = match crate::auth::extract_bearer_token_from_header(
         headers.get("Authorization").and_then(|h| h.to_str().ok()),
     ) {
@@ -53,12 +50,14 @@ pub async fn request_email_update(
                 .into_response();
         }
     };
+
     let auth_result = crate::auth::validate_bearer_token(&state.db, &token).await;
     let did = match auth_result {
         Ok(user) => user.did,
         Err(e) => return ApiError::from(e).into_response(),
     };
-    let user = match sqlx::query!("SELECT id, handle FROM users WHERE did = $1", did)
+
+    let user = match sqlx::query!("SELECT id, handle, email FROM users WHERE did = $1", did)
         .fetch_optional(&state.db)
         .await
     {
@@ -71,9 +70,12 @@ pub async fn request_email_update(
                 .into_response();
         }
     };
+
     let user_id = user.id;
     let handle = user.handle;
+    let current_email = user.email;
     let email = input.email.trim().to_lowercase();
+
     if !crate::api::validation::is_valid_email(&email) {
         return (
             StatusCode::BAD_REQUEST,
@@ -81,9 +83,19 @@ pub async fn request_email_update(
         )
             .into_response();
     }
-    let exists = sqlx::query!("SELECT 1 as one FROM users WHERE LOWER(email) = $1", email)
-        .fetch_optional(&state.db)
-        .await;
+
+    if current_email.as_ref().map(|e| e.to_lowercase()) == Some(email.clone()) {
+        return (StatusCode::OK, Json(json!({ "tokenRequired": false }))).into_response();
+    }
+
+    let exists = sqlx::query!(
+        "SELECT 1 as one FROM users WHERE LOWER(email) = $1 AND id != $2",
+        email,
+        user_id
+    )
+    .fetch_optional(&state.db)
+    .await;
+
     if let Ok(Some(_)) = exists {
         return (
             StatusCode::BAD_REQUEST,
@@ -91,33 +103,24 @@ pub async fn request_email_update(
         )
             .into_response();
     }
-    let code = generate_confirmation_code();
-    let expires_at = Utc::now() + Duration::minutes(10);
-    let update = sqlx::query!(
-        "UPDATE users SET email_pending_verification = $1, email_confirmation_code = $2, email_confirmation_code_expires_at = $3 WHERE id = $4",
-        email,
-        code,
-        expires_at,
-        user_id
+
+    if let Err(e) = crate::api::notification_prefs::request_channel_verification(
+        &state.db,
+        user_id,
+        "email",
+        &email,
+        Some(&handle),
     )
-    .execute(&state.db)
-    .await;
-    if let Err(e) = update {
-        error!("DB error setting email update code: {:?}", e);
+    .await
+    {
+        error!("Failed to request email verification: {}", 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_email_update(
-        &state.db, user_id, &email, &handle, &code, &hostname,
-    )
-    .await
-    {
-        warn!("Failed to enqueue email update notification: {:?}", e);
-    }
+
     info!("Email update requested for user {}", user_id);
     (StatusCode::OK, Json(json!({ "tokenRequired": true }))).into_response()
 }
@@ -149,6 +152,7 @@ pub async fn confirm_email(
         )
             .into_response();
     }
+
     let token = match crate::auth::extract_bearer_token_from_header(
         headers.get("Authorization").and_then(|h| h.to_str().ok()),
     ) {
@@ -161,20 +165,19 @@ pub async fn confirm_email(
                 .into_response();
         }
     };
+
     let auth_result = crate::auth::validate_bearer_token(&state.db, &token).await;
     let did = match auth_result {
         Ok(user) => user.did,
         Err(e) => return ApiError::from(e).into_response(),
     };
-    let user = match sqlx::query!(
-        "SELECT id, email_confirmation_code, email_confirmation_code_expires_at, email_pending_verification FROM users WHERE did = $1",
-        did
-    )
-    .fetch_optional(&state.db)
-    .await
+
+    let user_id = match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
+        .fetch_one(&state.db)
+        .await
     {
-        Ok(Some(row)) => row,
-        _ => {
+        Ok(id) => id,
+        Err(_) => {
             return (
                 StatusCode::INTERNAL_SERVER_ERROR,
                 Json(json!({"error": "InternalError"})),
@@ -182,25 +185,28 @@ pub async fn confirm_email(
                 .into_response();
         }
     };
-    let user_id = user.id;
-    let stored_code = user.email_confirmation_code;
-    let expires_at = user.email_confirmation_code_expires_at;
-    let email_pending_verification = user.email_pending_verification;
-    let email = input.email.trim().to_lowercase();
-    let confirmation_code = input.token.trim();
-    let (pending_email, saved_code, expiry) =
-        match (email_pending_verification, stored_code, expires_at) {
-            (Some(p), Some(c), Some(e)) => (p, c, e),
-            _ => {
-                return (
+
+    let verification = match sqlx::query!(
+        "SELECT code, pending_identifier, expires_at FROM channel_verifications WHERE user_id = $1 AND channel = 'email'",
+        user_id
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(row)) => row,
+        _ => {
+            return (
                 StatusCode::BAD_REQUEST,
-                Json(
-                    json!({"error": "InvalidRequest", "message": "No pending email update found"}),
-                ),
+                Json(json!({"error": "InvalidRequest", "message": "No pending email update found"})),
             )
                 .into_response();
-            }
-        };
+        }
+    };
+
+    let pending_email = verification.pending_identifier.unwrap_or_default();
+    let email = input.email.trim().to_lowercase();
+    let confirmation_code = input.token.trim();
+
     if pending_email != email {
         return (
             StatusCode::BAD_REQUEST,
@@ -208,27 +214,36 @@ pub async fn confirm_email(
         )
             .into_response();
     }
-    if saved_code != confirmation_code {
+
+    if verification.code != confirmation_code {
         return (
             StatusCode::BAD_REQUEST,
             Json(json!({"error": "InvalidToken", "message": "Invalid token"})),
         )
             .into_response();
     }
-    if Utc::now() > expiry {
+
+    if Utc::now() > verification.expires_at {
         return (
             StatusCode::BAD_REQUEST,
             Json(json!({"error": "ExpiredToken", "message": "Token has expired"})),
         )
             .into_response();
     }
+
+    let mut tx = match state.db.begin().await {
+        Ok(tx) => tx,
+        Err(_) => return ApiError::InternalError.into_response(),
+    };
+
     let update = sqlx::query!(
-        "UPDATE users SET email = $1, email_pending_verification = NULL, email_confirmation_code = NULL, email_confirmation_code_expires_at = NULL WHERE id = $2",
+        "UPDATE users SET email = $1, updated_at = NOW() WHERE id = $2",
         pending_email,
         user_id
     )
-    .execute(&state.db)
+    .execute(&mut *tx)
     .await;
+
     if let Err(e) = update {
         error!("DB error finalizing email update: {:?}", e);
         if e.as_database_error()
@@ -247,6 +262,22 @@ pub async fn confirm_email(
         )
             .into_response();
     }
+
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = 'email'",
+        user_id
+    )
+    .execute(&mut *tx)
+    .await
+    {
+        error!("Failed to delete verification record: {:?}", e);
+        return ApiError::InternalError.into_response();
+    }
+
+    if let Err(_) = tx.commit().await {
+        return ApiError::InternalError.into_response();
+    }
+
     info!("Email updated for user {}", user_id);
     (StatusCode::OK, Json(json!({}))).into_response()
 }
@@ -277,13 +308,15 @@ pub async fn update_email(
                 .into_response();
         }
     };
+
     let auth_result = crate::auth::validate_bearer_token(&state.db, &token).await;
     let did = match auth_result {
         Ok(user) => user.did,
         Err(e) => return ApiError::from(e).into_response(),
     };
+
     let user = match sqlx::query!(
-        "SELECT id, email, email_confirmation_code, email_confirmation_code_expires_at, email_pending_verification FROM users WHERE did = $1",
+        "SELECT id, email FROM users WHERE did = $1",
         did
     )
     .fetch_optional(&state.db)
@@ -298,12 +331,11 @@ pub async fn update_email(
                 .into_response();
         }
     };
+
     let user_id = user.id;
     let current_email = user.email;
-    let stored_code = user.email_confirmation_code;
-    let expires_at = user.email_confirmation_code_expires_at;
-    let email_pending_verification = user.email_pending_verification;
     let new_email = input.email.trim().to_lowercase();
+
     if !crate::api::validation::is_valid_email(&new_email) {
         return (
             StatusCode::BAD_REQUEST,
@@ -311,32 +343,34 @@ pub async fn update_email(
         )
             .into_response();
     }
+
     if let Some(ref current) = current_email
-        && new_email == current.to_lowercase() {
-            return (StatusCode::OK, Json(json!({}))).into_response();
-        }
-    let email_confirmed = stored_code.is_some() && email_pending_verification.is_some();
-    if email_confirmed {
+        && new_email == current.to_lowercase()
+    {
+        return (StatusCode::OK, Json(json!({}))).into_response();
+    }
+
+    let verification = sqlx::query!(
+        "SELECT code, pending_identifier, expires_at FROM channel_verifications WHERE user_id = $1 AND channel = 'email'",
+        user_id
+    )
+    .fetch_optional(&state.db)
+    .await
+    .unwrap_or(None);
+
+    if let Some(ver) = verification {
         let confirmation_token = match &input.token {
             Some(t) => t.trim(),
             None => {
                 return (
                     StatusCode::BAD_REQUEST,
-                    Json(json!({"error": "TokenRequired", "message": "Token required for confirmed accounts. Call requestEmailUpdate first."})),
-                )
-                    .into_response();
-            }
-        };
-        let pending_email = match email_pending_verification {
-            Some(p) => p,
-            None => {
-                return (
-                    StatusCode::BAD_REQUEST,
-                    Json(json!({"error": "InvalidRequest", "message": "No pending email update found"})),
+                    Json(json!({"error": "TokenRequired", "message": "Token required. Call requestEmailUpdate first."})),
                 )
                     .into_response();
             }
         };
+
+        let pending_email = ver.pending_identifier.unwrap_or_default();
         if pending_email.to_lowercase() != new_email {
             return (
                 StatusCode::BAD_REQUEST,
@@ -344,32 +378,24 @@ pub async fn update_email(
             )
                 .into_response();
         }
-        let saved_code = match stored_code {
-            Some(c) => c,
-            None => {
-                return (
-                    StatusCode::BAD_REQUEST,
-                    Json(json!({"error": "InvalidRequest", "message": "No pending email update found"})),
-                )
-                    .into_response();
-            }
-        };
-        if saved_code != confirmation_token {
+
+        if ver.code != confirmation_token {
             return (
                 StatusCode::BAD_REQUEST,
                 Json(json!({"error": "InvalidToken", "message": "Invalid token"})),
             )
                 .into_response();
         }
-        if let Some(exp) = expires_at
-            && Utc::now() > exp {
-                return (
-                    StatusCode::BAD_REQUEST,
-                    Json(json!({"error": "ExpiredToken", "message": "Token has expired"})),
-                )
-                    .into_response();
-            }
+
+        if Utc::now() > ver.expires_at {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "ExpiredToken", "message": "Token has expired"})),
+            )
+                .into_response();
+        }
     }
+
     let exists = sqlx::query!(
         "SELECT 1 as one FROM users WHERE LOWER(email) = $1 AND id != $2",
         new_email,
@@ -377,6 +403,7 @@ pub async fn update_email(
     )
     .fetch_optional(&state.db)
     .await;
+
     if let Ok(Some(_)) = exists {
         return (
             StatusCode::BAD_REQUEST,
@@ -384,43 +411,62 @@ pub async fn update_email(
         )
             .into_response();
     }
+
+    let mut tx = match state.db.begin().await {
+        Ok(tx) => tx,
+        Err(_) => return ApiError::InternalError.into_response(),
+    };
+
     let update = sqlx::query!(
-        r#"
-        UPDATE users
-        SET email = $1,
-            email_pending_verification = NULL,
-            email_confirmation_code = NULL,
-            email_confirmation_code_expires_at = NULL,
-            updated_at = NOW()
-        WHERE id = $2
-        "#,
+        "UPDATE users SET email = $1, updated_at = NOW() WHERE id = $2",
         new_email,
         user_id
     )
-    .execute(&state.db)
+    .execute(&mut *tx)
     .await;
-    match update {
-        Ok(_) => {
-            info!("Email updated for user {}", user_id);
-            (StatusCode::OK, Json(json!({}))).into_response()
-        }
-        Err(e) => {
-            error!("DB error finalizing email update: {:?}", e);
-            if e.as_database_error()
-                .map(|db_err| db_err.is_unique_violation())
-                .unwrap_or(false)
-            {
-                return (
-                    StatusCode::BAD_REQUEST,
-                    Json(json!({"error": "InvalidRequest", "message": "Email already in use"})),
-                )
-                    .into_response();
-            }
-            (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                Json(json!({"error": "InternalError"})),
+
+    if let Err(e) = update {
+        error!("DB error finalizing email update: {:?}", e);
+        if e.as_database_error()
+            .map(|db_err| db_err.is_unique_violation())
+            .unwrap_or(false)
+        {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidRequest", "message": "Email already in use"})),
             )
-                .into_response()
+                .into_response();
         }
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response();
     }
+
+    let _ = sqlx::query!(
+        "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = 'email'",
+        user_id
+    )
+    .execute(&mut *tx)
+    .await;
+
+    if let Err(_) = tx.commit().await {
+        return ApiError::InternalError.into_response();
+    }
+
+    match sqlx::query!(
+        "INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, 'email_auth_factor', $2) ON CONFLICT (user_id, name) DO UPDATE SET value_json = $2",
+        user_id,
+        json!(input.email_auth_factor.unwrap_or(false))
+    )
+    .execute(&state.db)
+    .await
+    {
+        Ok(_) => {}
+        Err(e) => warn!("Failed to update email_auth_factor preference: {}", e),
+    }
+
+    info!("Email updated for user {}", user_id);
+    (StatusCode::OK, Json(json!({}))).into_response()
 }

src/api/server/session.rs

@@ -475,8 +475,6 @@ pub async fn confirm_signup(
     let row = match sqlx::query!(
         r#"SELECT
             u.id, u.did, u.handle, u.email,
-            u.email_confirmation_code,
-            u.email_confirmation_code_expires_at,
             u.preferred_notification_channel as "channel: crate::notifications::NotificationChannel",
             k.key_bytes, k.encryption_version
         FROM users u
@@ -497,23 +495,35 @@ pub async fn confirm_signup(
             return ApiError::InternalError.into_response();
         }
     };
-    let stored_code = match &row.email_confirmation_code {
-        Some(code) => code,
-        None => {
+
+    let verification = match sqlx::query!(
+        "SELECT code, expires_at FROM channel_verifications WHERE user_id = $1 AND channel = 'email'",
+        row.id
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(v)) => v,
+        Ok(None) => {
             warn!("No verification code found for user: {}", input.did);
             return ApiError::InvalidRequest("No pending verification".into()).into_response();
         }
+        Err(e) => {
+            error!("Database error fetching verification: {:?}", e);
+            return ApiError::InternalError.into_response();
+        }
     };
-    if stored_code != &input.verification_code {
+
+    if verification.code != input.verification_code {
         warn!("Invalid verification code for user: {}", input.did);
         return ApiError::InvalidRequest("Invalid verification code".into()).into_response();
     }
-    if let Some(expires_at) = row.email_confirmation_code_expires_at
-        && expires_at < Utc::now() {
-            warn!("Verification code expired for user: {}", input.did);
-            return ApiError::ExpiredTokenMsg("Verification code has expired".into())
-                .into_response();
-        }
+    if verification.expires_at < Utc::now() {
+        warn!("Verification code expired for user: {}", input.did);
+        return ApiError::ExpiredTokenMsg("Verification code has expired".into())
+            .into_response();
+    }
+
     let key_bytes = match crate::config::decrypt_key(&row.key_bytes, row.encryption_version) {
         Ok(k) => k,
         Err(e) => {
@@ -528,7 +538,7 @@ pub async fn confirm_signup(
         crate::notifications::NotificationChannel::Signal => "signal_verified",
     };
     let update_query = format!(
-        "UPDATE users SET {} = TRUE, email_confirmation_code = NULL, email_confirmation_code_expires_at = NULL WHERE did = $1",
+        "UPDATE users SET {} = TRUE WHERE did = $1",
         verified_column
     );
     if let Err(e) = sqlx::query(&update_query)
@@ -539,6 +549,16 @@ pub async fn confirm_signup(
         error!("Failed to update verification status: {:?}", e);
         return ApiError::InternalError.into_response();
     }
+
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = 'email'",
+        row.id
+    )
+    .execute(&state.db)
+    .await {
+        error!("Failed to delete verification record: {:?}", e);
+    }
+
     let access_meta = match crate::auth::create_access_token_with_metadata(&row.did, &key_bytes) {
         Ok(m) => m,
         Err(e) => {
@@ -634,11 +654,20 @@ pub async fn resend_verification(
     }
     let verification_code = format!("{:06}", rand::random::<u32>() % 1_000_000);
     let code_expires_at = Utc::now() + chrono::Duration::minutes(30);
+
+    let email = row.email.clone();
+
     if let Err(e) = sqlx::query!(
-        "UPDATE users SET email_confirmation_code = $1, email_confirmation_code_expires_at = $2 WHERE did = $3",
+        r#"
+        INSERT INTO channel_verifications (user_id, channel, code, pending_identifier, expires_at)
+        VALUES ($1, 'email', $2, $3, $4)
+        ON CONFLICT (user_id, channel) DO UPDATE
+        SET code = $2, pending_identifier = $3, expires_at = $4, created_at = NOW()
+        "#,
+        row.id,
         verification_code,
-        code_expires_at,
-        input.did
+        email,
+        code_expires_at
     )
     .execute(&state.db)
     .await
@@ -648,7 +677,7 @@ pub async fn resend_verification(
     }
     let (channel_str, recipient) = match row.channel {
         crate::notifications::NotificationChannel::Email => {
-            ("email", row.email.clone().unwrap_or_default())
+            ("email", row.email.unwrap_or_default())
         }
         crate::notifications::NotificationChannel::Discord => {
             ("discord", row.discord_id.unwrap_or_default())

src/api/verification.rs

@@ -0,0 +1,191 @@
+use crate::auth::validate_bearer_token;
+use crate::state::AppState;
+use axum::{
+    Json,
+    extract::State,
+    http::{HeaderMap, StatusCode},
+    response::{IntoResponse, Response},
+};
+use chrono::Utc;
+use serde::Deserialize;
+use serde_json::json;
+use tracing::{error, info};
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ConfirmChannelVerificationInput {
+    pub channel: String,
+    pub code: String,
+}
+
+pub async fn confirm_channel_verification(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(input): Json<ConfirmChannelVerificationInput>,
+) -> 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", "message": "Authentication required"})),
+        )
+            .into_response(),
+    };
+    let user = match validate_bearer_token(&state.db, &token).await {
+        Ok(u) => u,
+        Err(_) => {
+            return (
+                StatusCode::UNAUTHORIZED,
+                Json(json!({"error": "AuthenticationFailed", "message": "Invalid token"})),
+            )
+                .into_response();
+        }
+    };
+
+    let user_id = match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", user.did)
+        .fetch_one(&state.db)
+        .await
+    {
+        Ok(id) => id,
+        Err(_) => return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "User not found"})),
+        )
+            .into_response(),
+    };
+
+    let channel_str = input.channel.as_str();
+    if !["email", "discord", "telegram", "signal"].contains(&channel_str) {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "Invalid channel"})),
+        )
+            .into_response();
+    }
+
+    let record = match sqlx::query!(
+        r#"
+        SELECT code, pending_identifier, expires_at FROM channel_verifications
+        WHERE user_id = $1 AND channel = $2::notification_channel
+        "#,
+        user_id,
+        channel_str as _
+    )
+    .fetch_optional(&state.db)
+    .await {
+        Ok(Some(r)) => r,
+        Ok(None) => return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "No pending verification found. Update notification preferences first."})),
+        )
+            .into_response(),
+        Err(e) => return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": format!("Database error: {}", e)})),
+        )
+            .into_response(),
+    };
+
+    let pending_identifier = match record.pending_identifier {
+        Some(p) => p,
+        None => return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "No pending identifier found"})),
+        )
+            .into_response(),
+    };
+
+    if record.expires_at < Utc::now() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "ExpiredToken", "message": "Verification code expired"})),
+        )
+            .into_response();
+    }
+
+    if record.code != input.code {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidCode", "message": "Invalid verification code"})),
+        )
+            .into_response();
+    }
+
+    let mut tx = match state.db.begin().await {
+        Ok(tx) => tx,
+        Err(_) => return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response(),
+    };
+
+    let update_result = match channel_str {
+        "email" => sqlx::query!(
+            "UPDATE users SET email = $1, updated_at = NOW() WHERE id = $2",
+            pending_identifier,
+            user_id
+        ).execute(&mut *tx).await,
+        "discord" => sqlx::query!(
+            "UPDATE users SET discord_id = $1, discord_verified = TRUE, updated_at = NOW() WHERE id = $2",
+            pending_identifier,
+            user_id
+        ).execute(&mut *tx).await,
+        "telegram" => sqlx::query!(
+            "UPDATE users SET telegram_username = $1, telegram_verified = TRUE, updated_at = NOW() WHERE id = $2",
+            pending_identifier,
+            user_id
+        ).execute(&mut *tx).await,
+        "signal" => sqlx::query!(
+            "UPDATE users SET signal_number = $1, signal_verified = TRUE, updated_at = NOW() WHERE id = $2",
+            pending_identifier,
+            user_id
+        ).execute(&mut *tx).await,
+        _ => unreachable!(),
+    };
+
+    if let Err(e) = update_result {
+        error!("Failed to update user channel: {:?}", e);
+        if channel_str == "email" && e.as_database_error().map(|db| db.is_unique_violation()).unwrap_or(false) {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "EmailTaken", "message": "Email already in use"})),
+            )
+                .into_response();
+        }
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "Failed to update channel"})),
+        )
+            .into_response();
+    }
+
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM channel_verifications WHERE user_id = $1 AND channel = $2::notification_channel",
+        user_id,
+        channel_str as _
+    )
+    .execute(&mut *tx)
+    .await {
+        error!("Failed to delete verification record: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response();
+    }
+
+    if let Err(_) = tx.commit().await {
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response();
+    }
+
+    info!(did = %user.did, channel = %channel_str, "Channel verified successfully");
+
+    Json(json!({"success": true})).into_response()
+}