Backups, adversarial migrations

src/api/backup.rs

@@ -0,0 +1,930 @@
+use crate::auth::BearerAuth;
+use crate::scheduled::generate_full_backup;
+use crate::state::AppState;
+use crate::storage::BackupStorage;
+use axum::{
+    Json,
+    extract::{Query, State},
+    http::StatusCode,
+    response::{IntoResponse, Response},
+};
+use cid::Cid;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::str::FromStr;
+use tracing::{error, info, warn};
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackupInfo {
+    pub id: String,
+    pub repo_rev: String,
+    pub repo_root_cid: String,
+    pub block_count: i32,
+    pub size_bytes: i64,
+    pub created_at: String,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListBackupsOutput {
+    pub backups: Vec<BackupInfo>,
+    pub backup_enabled: bool,
+}
+
+pub async fn list_backups(State(state): State<AppState>, auth: BearerAuth) -> Response {
+    let user = match sqlx::query!(
+        "SELECT id, backup_enabled FROM users WHERE did = $1",
+        auth.0.did
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(u)) => u,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "AccountNotFound", "message": "Account not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error fetching user: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    let backups = match sqlx::query!(
+        r#"
+        SELECT id, repo_rev, repo_root_cid, block_count, size_bytes, created_at
+        FROM account_backups
+        WHERE user_id = $1
+        ORDER BY created_at DESC
+        "#,
+        user.id
+    )
+    .fetch_all(&state.db)
+    .await
+    {
+        Ok(rows) => rows,
+        Err(e) => {
+            error!("DB error fetching backups: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    let backup_list: Vec<BackupInfo> = backups
+        .into_iter()
+        .map(|b| BackupInfo {
+            id: b.id.to_string(),
+            repo_rev: b.repo_rev,
+            repo_root_cid: b.repo_root_cid,
+            block_count: b.block_count,
+            size_bytes: b.size_bytes,
+            created_at: b.created_at.to_rfc3339(),
+        })
+        .collect();
+
+    (
+        StatusCode::OK,
+        Json(ListBackupsOutput {
+            backups: backup_list,
+            backup_enabled: user.backup_enabled,
+        }),
+    )
+        .into_response()
+}
+
+#[derive(Deserialize)]
+pub struct GetBackupQuery {
+    pub id: String,
+}
+
+pub async fn get_backup(
+    State(state): State<AppState>,
+    auth: BearerAuth,
+    Query(query): Query<GetBackupQuery>,
+) -> Response {
+    let backup_id = match uuid::Uuid::parse_str(&query.id) {
+        Ok(id) => id,
+        Err(_) => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidRequest", "message": "Invalid backup ID"})),
+            )
+                .into_response();
+        }
+    };
+
+    let backup = match sqlx::query!(
+        r#"
+        SELECT ab.storage_key, ab.repo_rev
+        FROM account_backups ab
+        JOIN users u ON u.id = ab.user_id
+        WHERE ab.id = $1 AND u.did = $2
+        "#,
+        backup_id,
+        auth.0.did
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(b)) => b,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "BackupNotFound", "message": "Backup not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error fetching backup: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    let backup_storage = match state.backup_storage.as_ref() {
+        Some(storage) => storage,
+        None => {
+            return (
+                StatusCode::SERVICE_UNAVAILABLE,
+                Json(
+                    json!({"error": "BackupsDisabled", "message": "Backup storage not configured"}),
+                ),
+            )
+                .into_response();
+        }
+    };
+
+    let car_bytes = match backup_storage.get_backup(&backup.storage_key).await {
+        Ok(bytes) => bytes,
+        Err(e) => {
+            error!("Failed to fetch backup from storage: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to retrieve backup"})),
+            )
+                .into_response();
+        }
+    };
+
+    (
+        StatusCode::OK,
+        [
+            (axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car"),
+            (
+                axum::http::header::CONTENT_DISPOSITION,
+                &format!("attachment; filename=\"{}.car\"", backup.repo_rev),
+            ),
+        ],
+        car_bytes,
+    )
+        .into_response()
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateBackupOutput {
+    pub id: String,
+    pub repo_rev: String,
+    pub size_bytes: i64,
+    pub block_count: i32,
+}
+
+pub async fn create_backup(State(state): State<AppState>, auth: BearerAuth) -> Response {
+    let backup_storage = match state.backup_storage.as_ref() {
+        Some(storage) => storage,
+        None => {
+            return (
+                StatusCode::SERVICE_UNAVAILABLE,
+                Json(
+                    json!({"error": "BackupsDisabled", "message": "Backup storage not configured"}),
+                ),
+            )
+                .into_response();
+        }
+    };
+
+    let user = match sqlx::query!(
+        r#"
+        SELECT u.id, u.did, u.backup_enabled, u.deactivated_at, r.repo_root_cid, r.repo_rev
+        FROM users u
+        JOIN repos r ON r.user_id = u.id
+        WHERE u.did = $1
+        "#,
+        auth.0.did
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(u)) => u,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "AccountNotFound", "message": "Account not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error fetching user: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    if user.deactivated_at.is_some() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "AccountDeactivated", "message": "Account is deactivated"})),
+        )
+            .into_response();
+    }
+
+    let repo_rev = match &user.repo_rev {
+        Some(rev) => rev.clone(),
+        None => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(
+                    json!({"error": "RepoNotReady", "message": "Repository not ready for backup"}),
+                ),
+            )
+                .into_response();
+        }
+    };
+
+    let head_cid = match Cid::from_str(&user.repo_root_cid) {
+        Ok(c) => c,
+        Err(_) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Invalid repo root CID"})),
+            )
+                .into_response();
+        }
+    };
+
+    let car_bytes = match generate_full_backup(&state.block_store, &head_cid).await {
+        Ok(bytes) => bytes,
+        Err(e) => {
+            error!("Failed to generate CAR: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to generate backup"})),
+            )
+                .into_response();
+        }
+    };
+
+    let block_count = crate::scheduled::count_car_blocks(&car_bytes);
+    let size_bytes = car_bytes.len() as i64;
+
+    let storage_key = match backup_storage
+        .put_backup(&user.did, &repo_rev, &car_bytes)
+        .await
+    {
+        Ok(key) => key,
+        Err(e) => {
+            error!("Failed to upload backup: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to store backup"})),
+            )
+                .into_response();
+        }
+    };
+
+    let backup_id = match sqlx::query_scalar!(
+        r#"
+        INSERT INTO account_backups (user_id, storage_key, repo_root_cid, repo_rev, block_count, size_bytes)
+        VALUES ($1, $2, $3, $4, $5, $6)
+        RETURNING id
+        "#,
+        user.id,
+        storage_key,
+        user.repo_root_cid,
+        repo_rev,
+        block_count,
+        size_bytes
+    )
+    .fetch_one(&state.db)
+    .await
+    {
+        Ok(id) => id,
+        Err(e) => {
+            error!("DB error inserting backup: {:?}", e);
+            if let Err(rollback_err) = backup_storage.delete_backup(&storage_key).await {
+                error!(
+                    storage_key = %storage_key,
+                    error = %rollback_err,
+                    "Failed to rollback orphaned backup from S3"
+                );
+            }
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to record backup"})),
+            )
+                .into_response();
+        }
+    };
+
+    info!(
+        did = %user.did,
+        rev = %repo_rev,
+        size_bytes,
+        "Created manual backup"
+    );
+
+    let retention = BackupStorage::retention_count();
+    if let Err(e) = cleanup_old_backups(&state.db, backup_storage, user.id, retention).await {
+        warn!(did = %user.did, error = %e, "Failed to cleanup old backups after manual backup");
+    }
+
+    (
+        StatusCode::OK,
+        Json(CreateBackupOutput {
+            id: backup_id.to_string(),
+            repo_rev,
+            size_bytes,
+            block_count,
+        }),
+    )
+        .into_response()
+}
+
+async fn cleanup_old_backups(
+    db: &sqlx::PgPool,
+    backup_storage: &BackupStorage,
+    user_id: uuid::Uuid,
+    retention_count: u32,
+) -> Result<(), String> {
+    let old_backups = sqlx::query!(
+        r#"
+        SELECT id, storage_key
+        FROM account_backups
+        WHERE user_id = $1
+        ORDER BY created_at DESC
+        OFFSET $2
+        "#,
+        user_id,
+        retention_count as i64
+    )
+    .fetch_all(db)
+    .await
+    .map_err(|e| format!("DB error fetching old backups: {}", e))?;
+
+    for backup in old_backups {
+        if let Err(e) = backup_storage.delete_backup(&backup.storage_key).await {
+            warn!(
+                storage_key = %backup.storage_key,
+                error = %e,
+                "Failed to delete old backup from storage, skipping DB cleanup to avoid orphan"
+            );
+            continue;
+        }
+
+        sqlx::query!("DELETE FROM account_backups WHERE id = $1", backup.id)
+            .execute(db)
+            .await
+            .map_err(|e| format!("Failed to delete old backup record: {}", e))?;
+    }
+
+    Ok(())
+}
+
+#[derive(Deserialize)]
+pub struct DeleteBackupQuery {
+    pub id: String,
+}
+
+pub async fn delete_backup(
+    State(state): State<AppState>,
+    auth: BearerAuth,
+    Query(query): Query<DeleteBackupQuery>,
+) -> Response {
+    let backup_id = match uuid::Uuid::parse_str(&query.id) {
+        Ok(id) => id,
+        Err(_) => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidRequest", "message": "Invalid backup ID"})),
+            )
+                .into_response();
+        }
+    };
+
+    let backup = match sqlx::query!(
+        r#"
+        SELECT ab.id, ab.storage_key, u.deactivated_at
+        FROM account_backups ab
+        JOIN users u ON u.id = ab.user_id
+        WHERE ab.id = $1 AND u.did = $2
+        "#,
+        backup_id,
+        auth.0.did
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(b)) => b,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "BackupNotFound", "message": "Backup not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error fetching backup: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    if backup.deactivated_at.is_some() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "AccountDeactivated", "message": "Account is deactivated"})),
+        )
+            .into_response();
+    }
+
+    if let Some(backup_storage) = state.backup_storage.as_ref()
+        && let Err(e) = backup_storage.delete_backup(&backup.storage_key).await
+    {
+        warn!(
+            storage_key = %backup.storage_key,
+            error = %e,
+            "Failed to delete backup from storage (continuing anyway)"
+        );
+    }
+
+    if let Err(e) = sqlx::query!("DELETE FROM account_backups WHERE id = $1", backup.id)
+        .execute(&state.db)
+        .await
+    {
+        error!("DB error deleting backup: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "Failed to delete backup"})),
+        )
+            .into_response();
+    }
+
+    info!(did = %auth.0.did, backup_id = %backup_id, "Deleted backup");
+
+    (StatusCode::OK, Json(json!({}))).into_response()
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SetBackupEnabledInput {
+    pub enabled: bool,
+}
+
+pub async fn set_backup_enabled(
+    State(state): State<AppState>,
+    auth: BearerAuth,
+    Json(input): Json<SetBackupEnabledInput>,
+) -> Response {
+    let user = match sqlx::query!(
+        "SELECT deactivated_at FROM users WHERE did = $1",
+        auth.0.did
+    )
+    .fetch_optional(&state.db)
+    .await
+    {
+        Ok(Some(u)) => u,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "AccountNotFound", "message": "Account not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error fetching user: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    if user.deactivated_at.is_some() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "AccountDeactivated", "message": "Account is deactivated"})),
+        )
+            .into_response();
+    }
+
+    if let Err(e) = sqlx::query!(
+        "UPDATE users SET backup_enabled = $1 WHERE did = $2",
+        input.enabled,
+        auth.0.did
+    )
+    .execute(&state.db)
+    .await
+    {
+        error!("DB error updating backup_enabled: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "Failed to update setting"})),
+        )
+            .into_response();
+    }
+
+    info!(did = %auth.0.did, enabled = input.enabled, "Updated backup_enabled setting");
+
+    (StatusCode::OK, Json(json!({"enabled": input.enabled}))).into_response()
+}
+
+pub async fn export_blobs(State(state): State<AppState>, auth: BearerAuth) -> Response {
+    let user = match sqlx::query!("SELECT id FROM users WHERE did = $1", auth.0.did)
+        .fetch_optional(&state.db)
+        .await
+    {
+        Ok(Some(u)) => u,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "AccountNotFound", "message": "Account not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error fetching user: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    let blobs = match sqlx::query!(
+        r#"
+        SELECT DISTINCT b.cid, b.storage_key, b.mime_type
+        FROM blobs b
+        JOIN record_blobs rb ON rb.blob_cid = b.cid
+        WHERE rb.repo_id = $1
+        "#,
+        user.id
+    )
+    .fetch_all(&state.db)
+    .await
+    {
+        Ok(rows) => rows,
+        Err(e) => {
+            error!("DB error fetching blobs: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Database error"})),
+            )
+                .into_response();
+        }
+    };
+
+    if blobs.is_empty() {
+        return (
+            StatusCode::OK,
+            [
+                (axum::http::header::CONTENT_TYPE, "application/zip"),
+                (
+                    axum::http::header::CONTENT_DISPOSITION,
+                    "attachment; filename=\"blobs.zip\"",
+                ),
+            ],
+            Vec::<u8>::new(),
+        )
+            .into_response();
+    }
+
+    let mut zip_buffer = std::io::Cursor::new(Vec::new());
+    {
+        let mut zip = zip::ZipWriter::new(&mut zip_buffer);
+
+        let options = zip::write::SimpleFileOptions::default()
+            .compression_method(zip::CompressionMethod::Deflated);
+
+        let mut exported: Vec<serde_json::Value> = Vec::new();
+        let mut skipped: Vec<serde_json::Value> = Vec::new();
+
+        for blob in &blobs {
+            let blob_data = match state.blob_store.get(&blob.storage_key).await {
+                Ok(data) => data,
+                Err(e) => {
+                    warn!(cid = %blob.cid, error = %e, "Failed to fetch blob, skipping");
+                    skipped.push(json!({
+                        "cid": blob.cid,
+                        "mimeType": blob.mime_type,
+                        "reason": "fetch_failed"
+                    }));
+                    continue;
+                }
+            };
+
+            let extension = mime_to_extension(&blob.mime_type);
+            let filename = format!("{}{}", blob.cid, extension);
+
+            if let Err(e) = zip.start_file(&filename, options) {
+                warn!(filename = %filename, error = %e, "Failed to start zip file entry");
+                skipped.push(json!({
+                    "cid": blob.cid,
+                    "mimeType": blob.mime_type,
+                    "reason": "zip_entry_failed"
+                }));
+                continue;
+            }
+
+            if let Err(e) = std::io::Write::write_all(&mut zip, &blob_data) {
+                warn!(filename = %filename, error = %e, "Failed to write blob to zip");
+                skipped.push(json!({
+                    "cid": blob.cid,
+                    "mimeType": blob.mime_type,
+                    "reason": "write_failed"
+                }));
+                continue;
+            }
+
+            exported.push(json!({
+                "cid": blob.cid,
+                "filename": filename,
+                "mimeType": blob.mime_type,
+                "sizeBytes": blob_data.len()
+            }));
+        }
+
+        let manifest = json!({
+            "exportedAt": chrono::Utc::now().to_rfc3339(),
+            "totalBlobs": blobs.len(),
+            "exportedCount": exported.len(),
+            "skippedCount": skipped.len(),
+            "exported": exported,
+            "skipped": skipped
+        });
+
+        if zip.start_file("manifest.json", options).is_ok() {
+            let _ = std::io::Write::write_all(
+                &mut zip,
+                serde_json::to_string_pretty(&manifest)
+                    .unwrap_or_else(|_| "{}".to_string())
+                    .as_bytes(),
+            );
+        }
+
+        if let Err(e) = zip.finish() {
+            error!("Failed to finish zip: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to create zip file"})),
+            )
+                .into_response();
+        }
+    }
+
+    let zip_bytes = zip_buffer.into_inner();
+
+    info!(did = %auth.0.did, blob_count = blobs.len(), size_bytes = zip_bytes.len(), "Exported blobs");
+
+    (
+        StatusCode::OK,
+        [
+            (axum::http::header::CONTENT_TYPE, "application/zip"),
+            (
+                axum::http::header::CONTENT_DISPOSITION,
+                "attachment; filename=\"blobs.zip\"",
+            ),
+        ],
+        zip_bytes,
+    )
+        .into_response()
+}
+
+fn mime_to_extension(mime_type: &str) -> &'static str {
+    match mime_type {
+        "application/font-sfnt" => ".otf",
+        "application/font-tdpfr" => ".pfr",
+        "application/font-woff" => ".woff",
+        "application/gzip" => ".gz",
+        "application/json" => ".json",
+        "application/json5" => ".json5",
+        "application/jsonml+json" => ".jsonml",
+        "application/octet-stream" => ".bin",
+        "application/pdf" => ".pdf",
+        "application/zip" => ".zip",
+        "audio/aac" => ".aac",
+        "audio/ac3" => ".ac3",
+        "audio/aiff" => ".aiff",
+        "audio/annodex" => ".axa",
+        "audio/audible" => ".aa",
+        "audio/basic" => ".au",
+        "audio/flac" => ".flac",
+        "audio/m4a" => ".m4a",
+        "audio/m4b" => ".m4b",
+        "audio/m4p" => ".m4p",
+        "audio/mid" => ".mid",
+        "audio/midi" => ".midi",
+        "audio/mp4" => ".mp4a",
+        "audio/mpeg" => ".mp3",
+        "audio/ogg" => ".ogg",
+        "audio/s3m" => ".s3m",
+        "audio/scpls" => ".pls",
+        "audio/silk" => ".sil",
+        "audio/vnd.audible.aax" => ".aax",
+        "audio/vnd.dece.audio" => ".uva",
+        "audio/vnd.digital-winds" => ".eol",
+        "audio/vnd.dlna.adts" => ".adt",
+        "audio/vnd.dra" => ".dra",
+        "audio/vnd.dts" => ".dts",
+        "audio/vnd.dts.hd" => ".dtshd",
+        "audio/vnd.lucent.voice" => ".lvp",
+        "audio/vnd.ms-playready.media.pya" => ".pya",
+        "audio/vnd.nuera.ecelp4800" => ".ecelp4800",
+        "audio/vnd.nuera.ecelp7470" => ".ecelp7470",
+        "audio/vnd.nuera.ecelp9600" => ".ecelp9600",
+        "audio/vnd.rip" => ".rip",
+        "audio/wav" => ".wav",
+        "audio/webm" => ".weba",
+        "audio/x-caf" => ".caf",
+        "audio/x-gsm" => ".gsm",
+        "audio/x-m4r" => ".m4r",
+        "audio/x-matroska" => ".mka",
+        "audio/x-mpegurl" => ".m3u",
+        "audio/x-ms-wax" => ".wax",
+        "audio/x-ms-wma" => ".wma",
+        "audio/x-pn-realaudio" => ".ra",
+        "audio/x-pn-realaudio-plugin" => ".rpm",
+        "audio/x-sd2" => ".sd2",
+        "audio/x-smd" => ".smd",
+        "audio/xm" => ".xm",
+        "font/collection" => ".ttc",
+        "font/ttf" => ".ttf",
+        "font/woff" => ".woff",
+        "font/woff2" => ".woff2",
+        "image/apng" => ".apng",
+        "image/avif" => ".avif",
+        "image/avif-sequence" => ".avifs",
+        "image/bmp" => ".bmp",
+        "image/cgm" => ".cgm",
+        "image/cis-cod" => ".cod",
+        "image/g3fax" => ".g3",
+        "image/gif" => ".gif",
+        "image/heic" => ".heic",
+        "image/heic-sequence" => ".heics",
+        "image/heif" => ".heif",
+        "image/heif-sequence" => ".heifs",
+        "image/ief" => ".ief",
+        "image/jp2" => ".jp2",
+        "image/jpeg" => ".jpg",
+        "image/jpm" => ".jpm",
+        "image/jpx" => ".jpf",
+        "image/jxl" => ".jxl",
+        "image/ktx" => ".ktx",
+        "image/pict" => ".pct",
+        "image/png" => ".png",
+        "image/prs.btif" => ".btif",
+        "image/qoi" => ".qoi",
+        "image/sgi" => ".sgi",
+        "image/svg+xml" => ".svg",
+        "image/tiff" => ".tiff",
+        "image/vnd.dece.graphic" => ".uvg",
+        "image/vnd.djvu" => ".djv",
+        "image/vnd.fastbidsheet" => ".fbs",
+        "image/vnd.fpx" => ".fpx",
+        "image/vnd.fst" => ".fst",
+        "image/vnd.fujixerox.edmics-mmr" => ".mmr",
+        "image/vnd.fujixerox.edmics-rlc" => ".rlc",
+        "image/vnd.ms-modi" => ".mdi",
+        "image/vnd.ms-photo" => ".wdp",
+        "image/vnd.net-fpx" => ".npx",
+        "image/vnd.radiance" => ".hdr",
+        "image/vnd.rn-realflash" => ".rf",
+        "image/vnd.wap.wbmp" => ".wbmp",
+        "image/vnd.xiff" => ".xif",
+        "image/webp" => ".webp",
+        "image/x-3ds" => ".3ds",
+        "image/x-adobe-dng" => ".dng",
+        "image/x-canon-cr2" => ".cr2",
+        "image/x-canon-cr3" => ".cr3",
+        "image/x-canon-crw" => ".crw",
+        "image/x-cmu-raster" => ".ras",
+        "image/x-cmx" => ".cmx",
+        "image/x-epson-erf" => ".erf",
+        "image/x-freehand" => ".fh",
+        "image/x-fuji-raf" => ".raf",
+        "image/x-icon" => ".ico",
+        "image/x-jg" => ".art",
+        "image/x-jng" => ".jng",
+        "image/x-kodak-dcr" => ".dcr",
+        "image/x-kodak-k25" => ".k25",
+        "image/x-kodak-kdc" => ".kdc",
+        "image/x-macpaint" => ".mac",
+        "image/x-minolta-mrw" => ".mrw",
+        "image/x-mrsid-image" => ".sid",
+        "image/x-nikon-nef" => ".nef",
+        "image/x-nikon-nrw" => ".nrw",
+        "image/x-olympus-orf" => ".orf",
+        "image/x-panasonic-rw" => ".raw",
+        "image/x-panasonic-rw2" => ".rw2",
+        "image/x-pentax-pef" => ".pef",
+        "image/x-portable-anymap" => ".pnm",
+        "image/x-portable-bitmap" => ".pbm",
+        "image/x-portable-graymap" => ".pgm",
+        "image/x-portable-pixmap" => ".ppm",
+        "image/x-qoi" => ".qoi",
+        "image/x-quicktime" => ".qti",
+        "image/x-rgb" => ".rgb",
+        "image/x-sigma-x3f" => ".x3f",
+        "image/x-sony-arw" => ".arw",
+        "image/x-sony-sr2" => ".sr2",
+        "image/x-sony-srf" => ".srf",
+        "image/x-tga" => ".tga",
+        "image/x-xbitmap" => ".xbm",
+        "image/x-xcf" => ".xcf",
+        "image/x-xpixmap" => ".xpm",
+        "image/x-xwindowdump" => ".xwd",
+        "model/gltf+json" => ".gltf",
+        "model/gltf-binary" => ".glb",
+        "model/iges" => ".igs",
+        "model/mesh" => ".msh",
+        "model/vnd.collada+xml" => ".dae",
+        "model/vnd.gdl" => ".gdl",
+        "model/vnd.gtw" => ".gtw",
+        "model/vnd.vtu" => ".vtu",
+        "model/vrml" => ".vrml",
+        "model/x3d+binary" => ".x3db",
+        "model/x3d+vrml" => ".x3dv",
+        "model/x3d+xml" => ".x3d",
+        "text/css" => ".css",
+        "text/html" => ".html",
+        "text/plain" => ".txt",
+        "video/3gpp" => ".3gp",
+        "video/3gpp2" => ".3g2",
+        "video/annodex" => ".axv",
+        "video/divx" => ".divx",
+        "video/h261" => ".h261",
+        "video/h263" => ".h263",
+        "video/h264" => ".h264",
+        "video/jpeg" => ".jpgv",
+        "video/jpm" => ".jpgm",
+        "video/mj2" => ".mj2",
+        "video/mp4" => ".mp4",
+        "video/mpeg" => ".mpg",
+        "video/ogg" => ".ogv",
+        "video/quicktime" => ".mov",
+        "video/vnd.dece.hd" => ".uvh",
+        "video/vnd.dece.mobile" => ".uvm",
+        "video/vnd.dece.pd" => ".uvp",
+        "video/vnd.dece.sd" => ".uvs",
+        "video/vnd.dece.video" => ".uvv",
+        "video/vnd.dlna.mpeg-tts" => ".ts",
+        "video/vnd.dvb.file" => ".dvb",
+        "video/vnd.fvt" => ".fvt",
+        "video/vnd.mpegurl" => ".m4u",
+        "video/vnd.ms-playready.media.pyv" => ".pyv",
+        "video/vnd.uvvu.mp4" => ".uvu",
+        "video/vnd.vivo" => ".viv",
+        "video/webm" => ".webm",
+        "video/x-dv" => ".dv",
+        "video/x-f4v" => ".f4v",
+        "video/x-fli" => ".fli",
+        "video/x-flv" => ".flv",
+        "video/x-ivf" => ".ivf",
+        "video/x-la-asf" => ".lsf",
+        "video/x-m4v" => ".m4v",
+        "video/x-matroska" => ".mkv",
+        "video/x-mng" => ".mng",
+        "video/x-ms-asf" => ".asf",
+        "video/x-ms-vob" => ".vob",
+        "video/x-ms-wm" => ".wm",
+        "video/x-ms-wmp" => ".wmp",
+        "video/x-ms-wmv" => ".wmv",
+        "video/x-ms-wmx" => ".wmx",
+        "video/x-ms-wvx" => ".wvx",
+        "video/x-msvideo" => ".avi",
+        "video/x-sgi-movie" => ".movie",
+        "video/x-smv" => ".smv",
+        _ => ".bin",
+    }
+}

src/api/mod.rs

@@ -1,6 +1,7 @@
 pub mod actor;
 pub mod admin;
 pub mod age_assurance;
+pub mod backup;
 pub mod delegation;
 pub mod error;
 pub mod identity;

src/api/notification_prefs.rs

@@ -182,15 +182,34 @@ pub async fn get_notification_history(
                 .into_response(),
         };
 
+    let sensitive_types = [
+        "email_verification",
+        "password_reset",
+        "email_update",
+        "two_factor_code",
+        "passkey_recovery",
+        "migration_verification",
+        "plc_operation",
+        "channel_verification",
+        "signup_verification",
+    ];
+
     let notifications = rows
         .iter()
-        .map(|row| NotificationHistoryEntry {
-            created_at: row.created_at.to_rfc3339(),
-            channel: row.channel.clone(),
-            comms_type: row.comms_type.clone(),
-            status: row.status.clone(),
-            subject: row.subject.clone(),
-            body: row.body.clone(),
+        .map(|row| {
+            let body = if sensitive_types.contains(&row.comms_type.as_str()) {
+                "[Code redacted for security]".to_string()
+            } else {
+                row.body.clone()
+            };
+            NotificationHistoryEntry {
+                created_at: row.created_at.to_rfc3339(),
+                channel: row.channel.clone(),
+                comms_type: row.comms_type.clone(),
+                status: row.status.clone(),
+                subject: row.subject.clone(),
+                body,
+            }
         })
         .collect();
 

src/api/repo/blob.rs

@@ -312,7 +312,7 @@ pub async fn list_missing_blobs(
         r#"
         SELECT rb.blob_cid, rb.record_uri
         FROM record_blobs rb
-        LEFT JOIN blobs b ON rb.blob_cid = b.cid AND b.created_by_user = rb.repo_id
+        LEFT JOIN blobs b ON rb.blob_cid = b.cid
         WHERE rb.repo_id = $1 AND b.cid IS NULL AND rb.blob_cid > $2
         ORDER BY rb.blob_cid
         LIMIT $3

src/api/repo/record/batch.rs

@@ -345,8 +345,9 @@ pub async fn apply_writes(
                 let rkey = rkey
                     .clone()
                     .unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
+                let record_ipld = crate::util::json_to_ipld(value);
                 let mut record_bytes = Vec::new();
-                if serde_ipld_dagcbor::to_writer(&mut record_bytes, value).is_err() {
+                if serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld).is_err() {
                     return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
                 }
                 let record_cid = match tracking_store.put(&record_bytes).await {
@@ -409,8 +410,9 @@ pub async fn apply_writes(
                     }
                 };
                 all_blob_cids.extend(extract_blob_cids(value));
+                let record_ipld = crate::util::json_to_ipld(value);
                 let mut record_bytes = Vec::new();
-                if serde_ipld_dagcbor::to_writer(&mut record_bytes, value).is_err() {
+                if serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld).is_err() {
                     return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
                 }
                 let record_cid = match tracking_store.put(&record_bytes).await {

src/api/repo/record/utils.rs

@@ -382,8 +382,9 @@ pub async fn create_record_internal(
     let commit = jacquard_repo::commit::Commit::from_cbor(&commit_bytes)
         .map_err(|e| format!("Failed to parse commit: {:?}", e))?;
     let mst = Mst::load(Arc::new(tracking_store.clone()), commit.data, None);
+    let record_ipld = crate::util::json_to_ipld(record);
     let mut record_bytes = Vec::new();
-    serde_ipld_dagcbor::to_writer(&mut record_bytes, record)
+    serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld)
         .map_err(|e| format!("Failed to serialize record: {:?}", e))?;
     let record_cid = tracking_store
         .put(&record_bytes)

src/api/repo/record/write.rs

@@ -297,8 +297,9 @@ pub async fn create_record(
     let rkey = input
         .rkey
         .unwrap_or_else(|| Tid::now(LimitedU32::MIN).to_string());
+    let record_ipld = crate::util::json_to_ipld(&input.record);
     let mut record_bytes = Vec::new();
-    if serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record).is_err() {
+    if serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld).is_err() {
         return (
             StatusCode::BAD_REQUEST,
             Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"})),
@@ -550,8 +551,9 @@ pub async fn put_record(
         }
     }
     let existing_cid = mst.get(&key).await.ok().flatten();
+    let record_ipld = crate::util::json_to_ipld(&input.record);
     let mut record_bytes = Vec::new();
-    if serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record).is_err() {
+    if serde_ipld_dagcbor::to_writer(&mut record_bytes, &record_ipld).is_err() {
         return (
             StatusCode::BAD_REQUEST,
             Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"})),

src/api/server/account_status.rs

@@ -567,7 +567,6 @@ pub async fn activate_account(
 #[serde(rename_all = "camelCase")]
 pub struct DeactivateAccountInput {
     pub delete_after: Option<String>,
-    pub migrating_to: Option<String>,
 }
 
 pub async fn deactivate_account(
@@ -618,62 +617,34 @@ pub async fn deactivate_account(
 
     let did = auth_user.did;
 
-    let migrating_to = if let Some(ref url) = input.migrating_to {
-        let url = url.trim().trim_end_matches('/');
-        if url.is_empty() || !did.starts_with("did:web:") {
-            None
-        } else {
-            if !url.starts_with("https://") {
-                return ApiError::InvalidRequest("migratingTo must start with https://".into())
-                    .into_response();
-            }
-            Some(url.to_string())
-        }
-    } else {
-        None
-    };
-
     let handle = sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", did)
         .fetch_optional(&state.db)
         .await
         .ok()
         .flatten();
 
-    let result = if let Some(ref pds_url) = migrating_to {
-        sqlx::query!(
-            "UPDATE users SET deactivated_at = NOW(), delete_after = $2, migrated_to_pds = $3, migrated_at = NOW() WHERE did = $1",
-            did,
-            delete_after,
-            pds_url
-        )
-        .execute(&state.db)
-        .await
-    } else {
-        sqlx::query!(
-            "UPDATE users SET deactivated_at = NOW(), delete_after = $2 WHERE did = $1",
-            did,
-            delete_after
-        )
-        .execute(&state.db)
-        .await
-    };
-
-    let status = if migrating_to.is_some() {
-        "migrated"
-    } else {
-        "deactivated"
-    };
+    let result = sqlx::query!(
+        "UPDATE users SET deactivated_at = NOW(), delete_after = $2 WHERE did = $1",
+        did,
+        delete_after
+    )
+    .execute(&state.db)
+    .await;
 
     match result {
         Ok(_) => {
             if let Some(ref h) = handle {
                 let _ = state.cache.delete(&format!("handle:{}", h)).await;
             }
-            if let Err(e) =
-                crate::api::repo::record::sequence_account_event(&state, &did, false, Some(status))
-                    .await
+            if let Err(e) = crate::api::repo::record::sequence_account_event(
+                &state,
+                &did,
+                false,
+                Some("deactivated"),
+            )
+            .await
             {
-                warn!("Failed to sequence account {} event: {}", status, e);
+                warn!("Failed to sequence account deactivated event: {}", e);
             }
             (StatusCode::OK, Json(json!({}))).into_response()
         }

src/api/server/email.rs

@@ -476,3 +476,57 @@ pub async fn update_email(
     info!("Email updated for user {}", user_id);
     (StatusCode::OK, Json(json!({}))).into_response()
 }
+
+#[derive(Deserialize)]
+pub struct CheckEmailVerifiedInput {
+    pub identifier: String,
+}
+
+pub async fn check_email_verified(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    Json(input): Json<CheckEmailVerifiedInput>,
+) -> Response {
+    let client_ip = crate::rate_limit::extract_client_ip(&headers, None);
+    if !state
+        .check_rate_limit(RateLimitKind::VerificationCheck, &client_ip)
+        .await
+    {
+        return (
+            StatusCode::TOO_MANY_REQUESTS,
+            Json(json!({
+                "error": "RateLimitExceeded",
+                "message": "Too many requests. Please try again later."
+            })),
+        )
+            .into_response();
+    }
+
+    let user = sqlx::query!(
+        "SELECT email_verified FROM users WHERE email = $1 OR handle = $1",
+        input.identifier
+    )
+    .fetch_optional(&state.db)
+    .await;
+
+    match user {
+        Ok(Some(row)) => (
+            StatusCode::OK,
+            Json(json!({ "verified": row.email_verified })),
+        )
+            .into_response(),
+        Ok(None) => (
+            StatusCode::NOT_FOUND,
+            Json(json!({ "error": "AccountNotFound", "message": "Account not found" })),
+        )
+            .into_response(),
+        Err(e) => {
+            error!("DB error checking email verified: {:?}", e);
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({ "error": "InternalError" })),
+            )
+                .into_response()
+        }
+    }
+}

src/api/server/migration.rs

@@ -6,238 +6,10 @@ use axum::{
     http::StatusCode,
     response::{IntoResponse, Response},
 };
-use chrono::{DateTime, Utc};
+use chrono::Utc;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 
-#[derive(Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct GetMigrationStatusOutput {
-    pub did: String,
-    pub did_type: String,
-    pub migrated: bool,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub migrated_to_pds: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub migrated_at: Option<DateTime<Utc>>,
-}
-
-pub async fn get_migration_status(
-    State(state): State<AppState>,
-    headers: axum::http::HeaderMap,
-) -> Response {
-    let extracted = match crate::auth::extract_auth_token_from_header(
-        headers.get("Authorization").and_then(|h| h.to_str().ok()),
-    ) {
-        Some(t) => t,
-        None => return ApiError::AuthenticationRequired.into_response(),
-    };
-    let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
-    let http_uri = format!(
-        "https://{}/xrpc/com.tranquil.account.getMigrationStatus",
-        std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
-    );
-    let auth_user = match crate::auth::validate_token_with_dpop(
-        &state.db,
-        &extracted.token,
-        extracted.is_dpop,
-        dpop_proof,
-        "GET",
-        &http_uri,
-        true,
-    )
-    .await
-    {
-        Ok(user) => user,
-        Err(e) => return ApiError::from(e).into_response(),
-    };
-    let user = match sqlx::query!(
-        "SELECT did, migrated_to_pds, migrated_at FROM users WHERE did = $1",
-        auth_user.did
-    )
-    .fetch_optional(&state.db)
-    .await
-    {
-        Ok(Some(row)) => row,
-        Ok(None) => return ApiError::AccountNotFound.into_response(),
-        Err(e) => {
-            tracing::error!("DB error getting migration status: {:?}", e);
-            return ApiError::InternalError.into_response();
-        }
-    };
-    let did_type = if user.did.starts_with("did:plc:") {
-        "plc"
-    } else if user.did.starts_with("did:web:") {
-        "web"
-    } else {
-        "unknown"
-    };
-    let migrated = user.migrated_to_pds.is_some();
-    (
-        StatusCode::OK,
-        Json(GetMigrationStatusOutput {
-            did: user.did,
-            did_type: did_type.to_string(),
-            migrated,
-            migrated_to_pds: user.migrated_to_pds,
-            migrated_at: user.migrated_at,
-        }),
-    )
-        .into_response()
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct UpdateMigrationForwardingInput {
-    pub pds_url: String,
-}
-
-#[derive(Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct UpdateMigrationForwardingOutput {
-    pub success: bool,
-    pub migrated_to_pds: String,
-    pub migrated_at: DateTime<Utc>,
-}
-
-pub async fn update_migration_forwarding(
-    State(state): State<AppState>,
-    headers: axum::http::HeaderMap,
-    Json(input): Json<UpdateMigrationForwardingInput>,
-) -> Response {
-    let extracted = match crate::auth::extract_auth_token_from_header(
-        headers.get("Authorization").and_then(|h| h.to_str().ok()),
-    ) {
-        Some(t) => t,
-        None => return ApiError::AuthenticationRequired.into_response(),
-    };
-    let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
-    let http_uri = format!(
-        "https://{}/xrpc/com.tranquil.account.updateMigrationForwarding",
-        std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
-    );
-    let auth_user = match crate::auth::validate_token_with_dpop(
-        &state.db,
-        &extracted.token,
-        extracted.is_dpop,
-        dpop_proof,
-        "POST",
-        &http_uri,
-        true,
-    )
-    .await
-    {
-        Ok(user) => user,
-        Err(e) => return ApiError::from(e).into_response(),
-    };
-    if !auth_user.did.starts_with("did:web:") {
-        return (
-            StatusCode::BAD_REQUEST,
-            Json(json!({
-                "error": "InvalidRequest",
-                "message": "Migration forwarding is only available for did:web accounts. did:plc accounts use PLC directory for identity updates."
-            })),
-        )
-            .into_response();
-    }
-    let pds_url = input.pds_url.trim();
-    if pds_url.is_empty() {
-        return ApiError::InvalidRequest("pds_url is required".into()).into_response();
-    }
-    if !pds_url.starts_with("https://") {
-        return ApiError::InvalidRequest("pds_url must start with https://".into()).into_response();
-    }
-    let pds_url_clean = pds_url.trim_end_matches('/');
-    let now = Utc::now();
-    let result = sqlx::query!(
-        "UPDATE users SET migrated_to_pds = $1, migrated_at = $2 WHERE did = $3",
-        pds_url_clean,
-        now,
-        auth_user.did
-    )
-    .execute(&state.db)
-    .await;
-    match result {
-        Ok(_) => {
-            tracing::info!(
-                "Updated migration forwarding for {} to {}",
-                auth_user.did,
-                pds_url_clean
-            );
-            (
-                StatusCode::OK,
-                Json(UpdateMigrationForwardingOutput {
-                    success: true,
-                    migrated_to_pds: pds_url_clean.to_string(),
-                    migrated_at: now,
-                }),
-            )
-                .into_response()
-        }
-        Err(e) => {
-            tracing::error!("DB error updating migration forwarding: {:?}", e);
-            ApiError::InternalError.into_response()
-        }
-    }
-}
-
-pub async fn clear_migration_forwarding(
-    State(state): State<AppState>,
-    headers: axum::http::HeaderMap,
-) -> Response {
-    let extracted = match crate::auth::extract_auth_token_from_header(
-        headers.get("Authorization").and_then(|h| h.to_str().ok()),
-    ) {
-        Some(t) => t,
-        None => return ApiError::AuthenticationRequired.into_response(),
-    };
-    let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
-    let http_uri = format!(
-        "https://{}/xrpc/com.tranquil.account.clearMigrationForwarding",
-        std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
-    );
-    let auth_user = match crate::auth::validate_token_with_dpop(
-        &state.db,
-        &extracted.token,
-        extracted.is_dpop,
-        dpop_proof,
-        "POST",
-        &http_uri,
-        true,
-    )
-    .await
-    {
-        Ok(user) => user,
-        Err(e) => return ApiError::from(e).into_response(),
-    };
-    if !auth_user.did.starts_with("did:web:") {
-        return (
-            StatusCode::BAD_REQUEST,
-            Json(json!({
-                "error": "InvalidRequest",
-                "message": "Migration forwarding is only available for did:web accounts"
-            })),
-        )
-            .into_response();
-    }
-    let result = sqlx::query!(
-        "UPDATE users SET migrated_to_pds = NULL, migrated_at = NULL WHERE did = $1",
-        auth_user.did
-    )
-    .execute(&state.db)
-    .await;
-    match result {
-        Ok(_) => {
-            tracing::info!("Cleared migration forwarding for {}", auth_user.did);
-            (StatusCode::OK, Json(json!({ "success": true }))).into_response()
-        }
-        Err(e) => {
-            tracing::error!("DB error clearing migration forwarding: {:?}", e);
-            ApiError::InternalError.into_response()
-        }
-    }
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct VerificationMethod {
@@ -275,7 +47,7 @@ pub async fn update_did_document(
     };
     let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
     let http_uri = format!(
-        "https://{}/xrpc/com.tranquil.account.updateDidDocument",
+        "https://{}/xrpc/_account.updateDidDocument",
         std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
     );
     let auth_user = match crate::auth::validate_token_with_dpop(
@@ -305,7 +77,7 @@ pub async fn update_did_document(
     }
 
     let user = match sqlx::query!(
-        "SELECT id, migrated_to_pds, handle FROM users WHERE did = $1",
+        "SELECT id, handle, deactivated_at FROM users WHERE did = $1",
         auth_user.did
     )
     .fetch_optional(&state.db)
@@ -319,15 +91,8 @@ pub async fn update_did_document(
         }
     };
 
-    if user.migrated_to_pds.is_none() {
-        return (
-            StatusCode::BAD_REQUEST,
-            Json(json!({
-                "error": "InvalidRequest",
-                "message": "DID document updates are only available for migrated accounts. Use the migration flow to migrate first."
-            })),
-        )
-            .into_response();
+    if user.deactivated_at.is_some() {
+        return ApiError::AccountDeactivated.into_response();
     }
 
     if let Some(ref methods) = input.verification_methods {
@@ -452,7 +217,7 @@ pub async fn get_did_document(
     };
     let dpop_proof = headers.get("DPoP").and_then(|h| h.to_str().ok());
     let http_uri = format!(
-        "https://{}/xrpc/com.tranquil.account.getDidDocument",
+        "https://{}/xrpc/_account.getDidDocument",
         std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string())
     );
     let auth_user = match crate::auth::validate_token_with_dpop(

src/api/server/mod.rs

@@ -22,14 +22,11 @@ pub use account_status::{
     request_account_delete,
 };
 pub use app_password::{create_app_password, list_app_passwords, revoke_app_password};
-pub use email::{confirm_email, request_email_update, update_email};
+pub use email::{check_email_verified, confirm_email, request_email_update, update_email};
 pub use invite::{create_invite_code, create_invite_codes, get_account_invite_codes};
 pub use logo::get_logo;
 pub use meta::{describe_server, health, robots_txt};
-pub use migration::{
-    clear_migration_forwarding, get_did_document, get_migration_status, update_did_document,
-    update_migration_forwarding,
-};
+pub use migration::{get_did_document, update_did_document};
 pub use passkey_account::{
     complete_passkey_setup, create_passkey_account, recover_passkey_account,
     request_passkey_recovery, start_passkey_registration_for_setup,

src/lib.rs

@@ -57,15 +57,15 @@ pub fn app(state: AppState) -> Router {
             get(api::server::get_session),
         )
         .route(
-            "/xrpc/com.tranquil.account.listSessions",
+            "/xrpc/_account.listSessions",
             get(api::server::list_sessions),
         )
         .route(
-            "/xrpc/com.tranquil.account.revokeSession",
+            "/xrpc/_account.revokeSession",
             post(api::server::revoke_session),
         )
         .route(
-            "/xrpc/com.tranquil.account.revokeAllSessions",
+            "/xrpc/_account.revokeAllSessions",
             post(api::server::revoke_all_sessions),
         )
         .route(
@@ -208,105 +208,94 @@ pub fn app(state: AppState) -> Router {
             post(api::server::reset_password),
         )
         .route(
-            "/xrpc/com.tranquil.account.changePassword",
+            "/xrpc/_account.changePassword",
             post(api::server::change_password),
         )
         .route(
-            "/xrpc/com.tranquil.account.removePassword",
+            "/xrpc/_account.removePassword",
             post(api::server::remove_password),
         )
         .route(
-            "/xrpc/com.tranquil.account.getPasswordStatus",
+            "/xrpc/_account.getPasswordStatus",
             get(api::server::get_password_status),
         )
         .route(
-            "/xrpc/com.tranquil.account.getReauthStatus",
+            "/xrpc/_account.getReauthStatus",
             get(api::server::get_reauth_status),
         )
         .route(
-            "/xrpc/com.tranquil.account.reauthPassword",
+            "/xrpc/_account.reauthPassword",
             post(api::server::reauth_password),
         )
+        .route("/xrpc/_account.reauthTotp", post(api::server::reauth_totp))
         .route(
-            "/xrpc/com.tranquil.account.reauthTotp",
-            post(api::server::reauth_totp),
-        )
-        .route(
-            "/xrpc/com.tranquil.account.reauthPasskeyStart",
+            "/xrpc/_account.reauthPasskeyStart",
             post(api::server::reauth_passkey_start),
         )
         .route(
-            "/xrpc/com.tranquil.account.reauthPasskeyFinish",
+            "/xrpc/_account.reauthPasskeyFinish",
             post(api::server::reauth_passkey_finish),
         )
         .route(
-            "/xrpc/com.tranquil.account.getLegacyLoginPreference",
+            "/xrpc/_account.getLegacyLoginPreference",
             get(api::server::get_legacy_login_preference),
         )
         .route(
-            "/xrpc/com.tranquil.account.updateLegacyLoginPreference",
+            "/xrpc/_account.updateLegacyLoginPreference",
             post(api::server::update_legacy_login_preference),
         )
         .route(
-            "/xrpc/com.tranquil.account.updateLocale",
+            "/xrpc/_account.updateLocale",
             post(api::server::update_locale),
         )
         .route(
-            "/xrpc/com.tranquil.account.listTrustedDevices",
+            "/xrpc/_account.listTrustedDevices",
             get(api::server::list_trusted_devices),
         )
         .route(
-            "/xrpc/com.tranquil.account.revokeTrustedDevice",
+            "/xrpc/_account.revokeTrustedDevice",
             post(api::server::revoke_trusted_device),
         )
         .route(
-            "/xrpc/com.tranquil.account.updateTrustedDevice",
+            "/xrpc/_account.updateTrustedDevice",
             post(api::server::update_trusted_device),
         )
         .route(
-            "/xrpc/com.tranquil.account.createPasskeyAccount",
+            "/xrpc/_account.createPasskeyAccount",
             post(api::server::create_passkey_account),
         )
         .route(
-            "/xrpc/com.tranquil.account.startPasskeyRegistrationForSetup",
+            "/xrpc/_account.startPasskeyRegistrationForSetup",
             post(api::server::start_passkey_registration_for_setup),
         )
         .route(
-            "/xrpc/com.tranquil.account.completePasskeySetup",
+            "/xrpc/_account.completePasskeySetup",
             post(api::server::complete_passkey_setup),
         )
         .route(
-            "/xrpc/com.tranquil.account.requestPasskeyRecovery",
+            "/xrpc/_account.requestPasskeyRecovery",
             post(api::server::request_passkey_recovery),
         )
         .route(
-            "/xrpc/com.tranquil.account.recoverPasskeyAccount",
+            "/xrpc/_account.recoverPasskeyAccount",
             post(api::server::recover_passkey_account),
         )
         .route(
-            "/xrpc/com.tranquil.account.getMigrationStatus",
-            get(api::server::get_migration_status),
-        )
-        .route(
-            "/xrpc/com.tranquil.account.updateMigrationForwarding",
-            post(api::server::update_migration_forwarding),
-        )
-        .route(
-            "/xrpc/com.tranquil.account.clearMigrationForwarding",
-            post(api::server::clear_migration_forwarding),
-        )
-        .route(
-            "/xrpc/com.tranquil.account.updateDidDocument",
+            "/xrpc/_account.updateDidDocument",
             post(api::server::update_did_document),
         )
         .route(
-            "/xrpc/com.tranquil.account.getDidDocument",
+            "/xrpc/_account.getDidDocument",
             get(api::server::get_did_document),
         )
         .route(
             "/xrpc/com.atproto.server.requestEmailUpdate",
             post(api::server::request_email_update),
         )
+        .route(
+            "/xrpc/_checkEmailVerified",
+            post(api::server::check_email_verified),
+        )
         .route(
             "/xrpc/com.atproto.server.confirmEmail",
             post(api::server::confirm_email),
@@ -432,15 +421,15 @@ pub fn app(state: AppState) -> Router {
             get(api::admin::get_invite_codes),
         )
         .route(
-            "/xrpc/com.tranquil.admin.getServerStats",
+            "/xrpc/_admin.getServerStats",
             get(api::admin::get_server_stats),
         )
         .route(
-            "/xrpc/com.tranquil.server.getConfig",
+            "/xrpc/_server.getConfig",
             get(api::admin::get_server_config),
         )
         .route(
-            "/xrpc/com.tranquil.admin.updateServerConfig",
+            "/xrpc/_admin.updateServerConfig",
             post(api::admin::update_server_config),
         )
         .route(
@@ -575,57 +564,72 @@ pub fn app(state: AppState) -> Router {
             post(api::temp::dereference_scope),
         )
         .route(
-            "/xrpc/com.tranquil.account.getNotificationPrefs",
+            "/xrpc/_account.getNotificationPrefs",
             get(api::notification_prefs::get_notification_prefs),
         )
         .route(
-            "/xrpc/com.tranquil.account.updateNotificationPrefs",
+            "/xrpc/_account.updateNotificationPrefs",
             post(api::notification_prefs::update_notification_prefs),
         )
         .route(
-            "/xrpc/com.tranquil.account.getNotificationHistory",
+            "/xrpc/_account.getNotificationHistory",
             get(api::notification_prefs::get_notification_history),
         )
         .route(
-            "/xrpc/com.tranquil.account.confirmChannelVerification",
+            "/xrpc/_account.confirmChannelVerification",
             post(api::verification::confirm_channel_verification),
         )
         .route(
-            "/xrpc/com.tranquil.account.verifyToken",
+            "/xrpc/_account.verifyToken",
             post(api::server::verify_token),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.listControllers",
+            "/xrpc/_delegation.listControllers",
             get(api::delegation::list_controllers),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.addController",
+            "/xrpc/_delegation.addController",
             post(api::delegation::add_controller),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.removeController",
+            "/xrpc/_delegation.removeController",
             post(api::delegation::remove_controller),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.updateControllerScopes",
+            "/xrpc/_delegation.updateControllerScopes",
             post(api::delegation::update_controller_scopes),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.listControlledAccounts",
+            "/xrpc/_delegation.listControlledAccounts",
             get(api::delegation::list_controlled_accounts),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.getAuditLog",
+            "/xrpc/_delegation.getAuditLog",
             get(api::delegation::get_audit_log),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.getScopePresets",
+            "/xrpc/_delegation.getScopePresets",
             get(api::delegation::get_scope_presets),
         )
         .route(
-            "/xrpc/com.tranquil.delegation.createDelegatedAccount",
+            "/xrpc/_delegation.createDelegatedAccount",
             post(api::delegation::create_delegated_account),
         )
+        .route("/xrpc/_backup.listBackups", get(api::backup::list_backups))
+        .route("/xrpc/_backup.getBackup", get(api::backup::get_backup))
+        .route(
+            "/xrpc/_backup.createBackup",
+            post(api::backup::create_backup),
+        )
+        .route(
+            "/xrpc/_backup.deleteBackup",
+            post(api::backup::delete_backup),
+        )
+        .route(
+            "/xrpc/_backup.setEnabled",
+            post(api::backup::set_backup_enabled),
+        )
+        .route("/xrpc/_backup.exportBlobs", get(api::backup::export_blobs))
         .route(
             "/xrpc/app.bsky.ageassurance.getState",
             get(api::age_assurance::get_state),

src/main.rs

@@ -7,7 +7,7 @@ use tranquil_pds::comms::{CommsService, DiscordSender, EmailSender, SignalSender
 use tranquil_pds::crawlers::{Crawlers, start_crawlers_service};
 use tranquil_pds::scheduled::{
     backfill_genesis_commit_blocks, backfill_record_blobs, backfill_repo_rev, backfill_user_blocks,
-    start_scheduled_tasks,
+    start_backup_tasks, start_scheduled_tasks,
 };
 use tranquil_pds::state::AppState;
 
@@ -83,6 +83,19 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
         None
     };
 
+    let backup_handle = if let Some(backup_storage) = state.backup_storage.clone() {
+        info!("Backup service enabled");
+        Some(tokio::spawn(start_backup_tasks(
+            state.db.clone(),
+            state.block_store.clone(),
+            backup_storage,
+            shutdown_rx.clone(),
+        )))
+    } else {
+        warn!("Backup service disabled (BACKUP_S3_BUCKET not set or BACKUP_ENABLED=false)");
+        None
+    };
+
     let scheduled_handle = tokio::spawn(start_scheduled_tasks(
         state.db.clone(),
         state.blob_store.clone(),
@@ -117,6 +130,10 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
         handle.await.ok();
     }
 
+    if let Some(handle) = backup_handle {
+        handle.await.ok();
+    }
+
     scheduled_handle.await.ok();
 
     if let Err(e) = server_result {

src/rate_limit.rs

@@ -32,6 +32,7 @@ pub struct RateLimiters {
     pub totp_verify: Arc<KeyedRateLimiter>,
     pub handle_update: Arc<KeyedRateLimiter>,
     pub handle_update_daily: Arc<KeyedRateLimiter>,
+    pub verification_check: Arc<KeyedRateLimiter>,
 }
 
 impl Default for RateLimiters {
@@ -91,6 +92,9 @@ impl RateLimiters {
                     .unwrap()
                     .allow_burst(NonZeroU32::new(50).unwrap()),
             )),
+            verification_check: Arc::new(RateLimiter::keyed(Quota::per_minute(
+                NonZeroU32::new(60).unwrap(),
+            ))),
         }
     }
 

src/scheduled.rs

@@ -11,7 +11,8 @@ use tokio::time::interval;
 use tracing::{debug, error, info, warn};
 
 use crate::repo::PostgresBlockStore;
-use crate::storage::BlobStorage;
+use crate::storage::{BackupStorage, BlobStorage};
+use crate::sync::car::encode_car_header;
 
 pub async fn backfill_genesis_commit_blocks(db: &PgPool, block_store: PostgresBlockStore) {
     let broken_genesis_commits = match sqlx::query!(
@@ -563,3 +564,312 @@ async fn delete_account_data(
 
     Ok(())
 }
+
+pub async fn start_backup_tasks(
+    db: PgPool,
+    block_store: PostgresBlockStore,
+    backup_storage: Arc<BackupStorage>,
+    mut shutdown_rx: watch::Receiver<bool>,
+) {
+    let backup_interval = Duration::from_secs(BackupStorage::interval_secs());
+
+    info!(
+        interval_secs = backup_interval.as_secs(),
+        retention_count = BackupStorage::retention_count(),
+        "Starting backup service"
+    );
+
+    let mut ticker = interval(backup_interval);
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+    loop {
+        tokio::select! {
+            _ = shutdown_rx.changed() => {
+                if *shutdown_rx.borrow() {
+                    info!("Backup service shutting down");
+                    break;
+                }
+            }
+            _ = ticker.tick() => {
+                if let Err(e) = process_scheduled_backups(&db, &block_store, &backup_storage).await {
+                    error!("Error processing scheduled backups: {}", e);
+                }
+            }
+        }
+    }
+}
+
+async fn process_scheduled_backups(
+    db: &PgPool,
+    block_store: &PostgresBlockStore,
+    backup_storage: &BackupStorage,
+) -> Result<(), String> {
+    let backup_interval_secs = BackupStorage::interval_secs() as i64;
+    let retention_count = BackupStorage::retention_count();
+
+    let users_needing_backup = sqlx::query!(
+        r#"
+        SELECT u.id as user_id, u.did, r.repo_root_cid, r.repo_rev
+        FROM users u
+        JOIN repos r ON r.user_id = u.id
+        WHERE u.backup_enabled = true
+          AND u.deactivated_at IS NULL
+          AND (
+            NOT EXISTS (
+              SELECT 1 FROM account_backups ab WHERE ab.user_id = u.id
+            )
+            OR (
+              SELECT MAX(ab.created_at) FROM account_backups ab WHERE ab.user_id = u.id
+            ) < NOW() - make_interval(secs => $1)
+          )
+        LIMIT 50
+        "#,
+        backup_interval_secs as f64
+    )
+    .fetch_all(db)
+    .await
+    .map_err(|e| format!("DB error fetching users for backup: {}", e))?;
+
+    if users_needing_backup.is_empty() {
+        debug!("No accounts need backup");
+        return Ok(());
+    }
+
+    info!(
+        count = users_needing_backup.len(),
+        "Processing scheduled backups"
+    );
+
+    for user in users_needing_backup {
+        let repo_root_cid = user.repo_root_cid.clone();
+
+        let repo_rev = match &user.repo_rev {
+            Some(rev) => rev.clone(),
+            None => {
+                warn!(did = %user.did, "User has no repo_rev, skipping backup");
+                continue;
+            }
+        };
+
+        let head_cid = match Cid::from_str(&repo_root_cid) {
+            Ok(c) => c,
+            Err(e) => {
+                warn!(did = %user.did, error = %e, "Invalid repo_root_cid, skipping backup");
+                continue;
+            }
+        };
+
+        let car_result = generate_full_backup(block_store, &head_cid).await;
+        let car_bytes = match car_result {
+            Ok(bytes) => bytes,
+            Err(e) => {
+                warn!(did = %user.did, error = %e, "Failed to generate CAR for backup");
+                continue;
+            }
+        };
+
+        let block_count = count_car_blocks(&car_bytes);
+        let size_bytes = car_bytes.len() as i64;
+
+        let storage_key = match backup_storage
+            .put_backup(&user.did, &repo_rev, &car_bytes)
+            .await
+        {
+            Ok(key) => key,
+            Err(e) => {
+                warn!(did = %user.did, error = %e, "Failed to upload backup to storage");
+                continue;
+            }
+        };
+
+        if let Err(e) = sqlx::query!(
+            r#"
+            INSERT INTO account_backups (user_id, storage_key, repo_root_cid, repo_rev, block_count, size_bytes)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            "#,
+            user.user_id,
+            storage_key,
+            repo_root_cid,
+            repo_rev,
+            block_count,
+            size_bytes
+        )
+        .execute(db)
+        .await
+        {
+            warn!(did = %user.did, error = %e, "Failed to insert backup record, rolling back S3 upload");
+            if let Err(rollback_err) = backup_storage.delete_backup(&storage_key).await {
+                error!(
+                    did = %user.did,
+                    storage_key = %storage_key,
+                    error = %rollback_err,
+                    "Failed to rollback orphaned backup from S3"
+                );
+            }
+            continue;
+        }
+
+        info!(
+            did = %user.did,
+            rev = %repo_rev,
+            size_bytes,
+            block_count,
+            "Created backup"
+        );
+
+        if let Err(e) = cleanup_old_backups(db, backup_storage, user.user_id, retention_count).await
+        {
+            warn!(did = %user.did, error = %e, "Failed to cleanup old backups");
+        }
+    }
+
+    Ok(())
+}
+
+pub async fn generate_repo_car(
+    block_store: &PostgresBlockStore,
+    head_cid: &Cid,
+) -> Result<Vec<u8>, String> {
+    use jacquard_repo::storage::BlockStore;
+    use std::io::Write;
+
+    let mut car_bytes =
+        encode_car_header(head_cid).map_err(|e| format!("Failed to encode CAR header: {}", e))?;
+
+    let mut stack = vec![*head_cid];
+    let mut visited = std::collections::HashSet::new();
+
+    while let Some(cid) = stack.pop() {
+        if visited.contains(&cid) {
+            continue;
+        }
+        visited.insert(cid);
+
+        if let Ok(Some(block)) = block_store.get(&cid).await {
+            let cid_bytes = cid.to_bytes();
+            let total_len = cid_bytes.len() + block.len();
+            let mut writer = Vec::new();
+            crate::sync::car::write_varint(&mut writer, total_len as u64)
+                .expect("Writing to Vec<u8> should never fail");
+            writer
+                .write_all(&cid_bytes)
+                .expect("Writing to Vec<u8> should never fail");
+            writer
+                .write_all(&block)
+                .expect("Writing to Vec<u8> should never fail");
+            car_bytes.extend_from_slice(&writer);
+
+            if let Ok(value) = serde_ipld_dagcbor::from_slice::<Ipld>(&block) {
+                extract_links(&value, &mut stack);
+            }
+        }
+    }
+
+    Ok(car_bytes)
+}
+
+pub async fn generate_full_backup(
+    block_store: &PostgresBlockStore,
+    head_cid: &Cid,
+) -> Result<Vec<u8>, String> {
+    generate_repo_car(block_store, head_cid).await
+}
+
+fn extract_links(value: &Ipld, stack: &mut Vec<Cid>) {
+    match value {
+        Ipld::Link(cid) => {
+            stack.push(*cid);
+        }
+        Ipld::Map(map) => {
+            for v in map.values() {
+                extract_links(v, stack);
+            }
+        }
+        Ipld::List(arr) => {
+            for v in arr {
+                extract_links(v, stack);
+            }
+        }
+        _ => {}
+    }
+}
+
+pub fn count_car_blocks(car_bytes: &[u8]) -> i32 {
+    let mut count = 0;
+    let mut pos = 0;
+
+    if let Some((header_len, header_varint_len)) = read_varint(&car_bytes[pos..]) {
+        pos += header_varint_len + header_len as usize;
+    } else {
+        return 0;
+    }
+
+    while pos < car_bytes.len() {
+        if let Some((block_len, varint_len)) = read_varint(&car_bytes[pos..]) {
+            pos += varint_len + block_len as usize;
+            count += 1;
+        } else {
+            break;
+        }
+    }
+
+    count
+}
+
+fn read_varint(data: &[u8]) -> Option<(u64, usize)> {
+    let mut value: u64 = 0;
+    let mut shift = 0;
+    let mut pos = 0;
+
+    while pos < data.len() && pos < 10 {
+        let byte = data[pos];
+        value |= ((byte & 0x7f) as u64) << shift;
+        pos += 1;
+        if byte & 0x80 == 0 {
+            return Some((value, pos));
+        }
+        shift += 7;
+    }
+
+    None
+}
+
+async fn cleanup_old_backups(
+    db: &PgPool,
+    backup_storage: &BackupStorage,
+    user_id: uuid::Uuid,
+    retention_count: u32,
+) -> Result<(), String> {
+    let old_backups = sqlx::query!(
+        r#"
+        SELECT id, storage_key
+        FROM account_backups
+        WHERE user_id = $1
+        ORDER BY created_at DESC
+        OFFSET $2
+        "#,
+        user_id,
+        retention_count as i64
+    )
+    .fetch_all(db)
+    .await
+    .map_err(|e| format!("DB error fetching old backups: {}", e))?;
+
+    for backup in old_backups {
+        if let Err(e) = backup_storage.delete_backup(&backup.storage_key).await {
+            warn!(
+                storage_key = %backup.storage_key,
+                error = %e,
+                "Failed to delete old backup from storage, skipping DB cleanup to avoid orphan"
+            );
+            continue;
+        }
+
+        sqlx::query!("DELETE FROM account_backups WHERE id = $1", backup.id)
+            .execute(db)
+            .await
+            .map_err(|e| format!("Failed to delete old backup record: {}", e))?;
+    }
+
+    Ok(())
+}

src/state.rs

@@ -4,7 +4,7 @@ use crate::circuit_breaker::CircuitBreakers;
 use crate::config::AuthConfig;
 use crate::rate_limit::RateLimiters;
 use crate::repo::PostgresBlockStore;
-use crate::storage::{BlobStorage, S3BlobStorage};
+use crate::storage::{BackupStorage, BlobStorage, S3BlobStorage};
 use crate::sync::firehose::SequencedEvent;
 use sqlx::PgPool;
 use std::error::Error;
@@ -16,6 +16,7 @@ pub struct AppState {
     pub db: PgPool,
     pub block_store: PostgresBlockStore,
     pub blob_store: Arc<dyn BlobStorage>,
+    pub backup_storage: Option<Arc<BackupStorage>>,
     pub firehose_tx: broadcast::Sender<SequencedEvent>,
     pub rate_limiters: Arc<RateLimiters>,
     pub circuit_breakers: Arc<CircuitBreakers>,
@@ -39,6 +40,7 @@ pub enum RateLimitKind {
     TotpVerify,
     HandleUpdate,
     HandleUpdateDaily,
+    VerificationCheck,
 }
 
 impl RateLimitKind {
@@ -58,6 +60,7 @@ impl RateLimitKind {
             Self::TotpVerify => "totp_verify",
             Self::HandleUpdate => "handle_update",
             Self::HandleUpdateDaily => "handle_update_daily",
+            Self::VerificationCheck => "verification_check",
         }
     }
 
@@ -77,6 +80,7 @@ impl RateLimitKind {
             Self::TotpVerify => (5, 300_000),
             Self::HandleUpdate => (10, 300_000),
             Self::HandleUpdateDaily => (50, 86_400_000),
+            Self::VerificationCheck => (60, 60_000),
         }
     }
 }
@@ -131,6 +135,7 @@ impl AppState {
 
         let block_store = PostgresBlockStore::new(db.clone());
         let blob_store = S3BlobStorage::new().await;
+        let backup_storage = BackupStorage::new().await.map(Arc::new);
 
         let firehose_buffer_size: usize = std::env::var("FIREHOSE_BUFFER_SIZE")
             .ok()
@@ -147,6 +152,7 @@ impl AppState {
             db,
             block_store,
             blob_store: Arc::new(blob_store),
+            backup_storage,
             firehose_tx,
             rate_limiters,
             circuit_breakers,
@@ -199,6 +205,7 @@ impl AppState {
             RateLimitKind::TotpVerify => &self.rate_limiters.totp_verify,
             RateLimitKind::HandleUpdate => &self.rate_limiters.handle_update,
             RateLimitKind::HandleUpdateDaily => &self.rate_limiters.handle_update_daily,
+            RateLimitKind::VerificationCheck => &self.rate_limiters.verification_check,
         };
 
         let ok = limiter.check_key(&client_ip.to_string()).is_ok();

src/storage/mod.rs

@@ -32,29 +32,132 @@ pub struct S3BlobStorage {
 
 impl S3BlobStorage {
     pub async fn new() -> Self {
-        let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
-
-        let config = aws_config::defaults(BehaviorVersion::latest())
-            .region(region_provider)
-            .load()
-            .await;
-
         let bucket = std::env::var("S3_BUCKET").expect("S3_BUCKET must be set");
-
-        let client = if let Ok(endpoint) = std::env::var("S3_ENDPOINT") {
-            let s3_config = aws_sdk_s3::config::Builder::from(&config)
-                .endpoint_url(endpoint)
-                .force_path_style(true)
-                .build();
-            Client::from_conf(s3_config)
-        } else {
-            Client::new(&config)
-        };
-
+        let client = create_s3_client().await;
         Self { client, bucket }
     }
 }
 
+async fn create_s3_client() -> Client {
+    let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
+
+    let config = aws_config::defaults(BehaviorVersion::latest())
+        .region(region_provider)
+        .load()
+        .await;
+
+    if let Ok(endpoint) = std::env::var("S3_ENDPOINT") {
+        let s3_config = aws_sdk_s3::config::Builder::from(&config)
+            .endpoint_url(endpoint)
+            .force_path_style(true)
+            .build();
+        Client::from_conf(s3_config)
+    } else {
+        Client::new(&config)
+    }
+}
+
+pub struct BackupStorage {
+    client: Client,
+    bucket: String,
+}
+
+impl BackupStorage {
+    pub async fn new() -> Option<Self> {
+        let backup_enabled = std::env::var("BACKUP_ENABLED")
+            .map(|v| v != "false" && v != "0")
+            .unwrap_or(true);
+
+        if !backup_enabled {
+            return None;
+        }
+
+        let bucket = std::env::var("BACKUP_S3_BUCKET").ok()?;
+        let client = create_s3_client().await;
+        Some(Self { client, bucket })
+    }
+
+    pub fn retention_count() -> u32 {
+        std::env::var("BACKUP_RETENTION_COUNT")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(7)
+    }
+
+    pub fn interval_secs() -> u64 {
+        std::env::var("BACKUP_INTERVAL_SECS")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(86400)
+    }
+
+    pub async fn put_backup(
+        &self,
+        did: &str,
+        rev: &str,
+        data: &[u8],
+    ) -> Result<String, StorageError> {
+        let key = format!("{}/{}.car", did, rev);
+        self.client
+            .put_object()
+            .bucket(&self.bucket)
+            .key(&key)
+            .body(ByteStream::from(Bytes::copy_from_slice(data)))
+            .send()
+            .await
+            .map_err(|e| {
+                crate::metrics::record_s3_operation("backup_put", "error");
+                StorageError::S3(e.to_string())
+            })?;
+
+        crate::metrics::record_s3_operation("backup_put", "success");
+        Ok(key)
+    }
+
+    pub async fn get_backup(&self, storage_key: &str) -> Result<Bytes, StorageError> {
+        let resp = self
+            .client
+            .get_object()
+            .bucket(&self.bucket)
+            .key(storage_key)
+            .send()
+            .await
+            .map_err(|e| {
+                crate::metrics::record_s3_operation("backup_get", "error");
+                StorageError::S3(e.to_string())
+            })?;
+
+        let data = resp
+            .body
+            .collect()
+            .await
+            .map_err(|e| {
+                crate::metrics::record_s3_operation("backup_get", "error");
+                StorageError::S3(e.to_string())
+            })?
+            .into_bytes();
+
+        crate::metrics::record_s3_operation("backup_get", "success");
+        Ok(data)
+    }
+
+    pub async fn delete_backup(&self, storage_key: &str) -> Result<(), StorageError> {
+        self.client
+            .delete_object()
+            .bucket(&self.bucket)
+            .key(storage_key)
+            .send()
+            .await
+            .map_err(|e| {
+                crate::metrics::record_s3_operation("backup_delete", "error");
+                StorageError::S3(e.to_string())
+            })?;
+
+        crate::metrics::record_s3_operation("backup_delete", "success");
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl BlobStorage for S3BlobStorage {
     async fn put(&self, key: &str, data: &[u8]) -> Result<(), StorageError> {

src/sync/import.rs

@@ -77,19 +77,30 @@ pub fn find_blob_refs_ipld(value: &Ipld, depth: usize) -> Vec<BlobRef> {
         Ipld::Map(obj) => {
             if let Some(Ipld::String(type_str)) = obj.get("$type")
                 && type_str == "blob"
-                && let Some(Ipld::Link(link_cid)) = obj.get("ref")
             {
-                let mime = obj.get("mimeType").and_then(|v| {
-                    if let Ipld::String(s) = v {
-                        Some(s.clone())
-                    } else {
-                        None
-                    }
-                });
-                return vec![BlobRef {
-                    cid: link_cid.to_string(),
-                    mime_type: mime,
-                }];
+                let cid_str = if let Some(Ipld::Link(link_cid)) = obj.get("ref") {
+                    Some(link_cid.to_string())
+                } else if let Some(Ipld::Map(ref_obj)) = obj.get("ref")
+                    && let Some(Ipld::String(link)) = ref_obj.get("$link")
+                {
+                    Some(link.clone())
+                } else {
+                    None
+                };
+
+                if let Some(cid) = cid_str {
+                    let mime = obj.get("mimeType").and_then(|v| {
+                        if let Ipld::String(s) = v {
+                            Some(s.clone())
+                        } else {
+                            None
+                        }
+                    });
+                    return vec![BlobRef {
+                        cid,
+                        mime_type: mime,
+                    }];
+                }
             }
             obj.values()
                 .flat_map(|v| find_blob_refs_ipld(v, depth + 1))

src/util.rs

@@ -1,6 +1,11 @@
 use axum::http::HeaderMap;
+use cid::Cid;
+use ipld_core::ipld::Ipld;
 use rand::Rng;
+use serde_json::Value as JsonValue;
 use sqlx::PgPool;
+use std::collections::BTreeMap;
+use std::str::FromStr;
 use std::sync::OnceLock;
 use uuid::Uuid;
 
@@ -150,6 +155,37 @@ pub fn build_full_url(path: &str) -> String {
     format!("{}{}", pds_public_url(), path)
 }
 
+pub fn json_to_ipld(value: &JsonValue) -> Ipld {
+    match value {
+        JsonValue::Null => Ipld::Null,
+        JsonValue::Bool(b) => Ipld::Bool(*b),
+        JsonValue::Number(n) => {
+            if let Some(i) = n.as_i64() {
+                Ipld::Integer(i as i128)
+            } else if let Some(f) = n.as_f64() {
+                Ipld::Float(f)
+            } else {
+                Ipld::Null
+            }
+        }
+        JsonValue::String(s) => Ipld::String(s.clone()),
+        JsonValue::Array(arr) => Ipld::List(arr.iter().map(json_to_ipld).collect()),
+        JsonValue::Object(obj) => {
+            if let Some(JsonValue::String(link)) = obj.get("$link")
+                && obj.len() == 1
+                && let Ok(cid) = Cid::from_str(link)
+            {
+                return Ipld::Link(cid);
+            }
+            let map: BTreeMap<String, Ipld> = obj
+                .iter()
+                .map(|(k, v)| (k.clone(), json_to_ipld(v)))
+                .collect();
+            Ipld::Map(map)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -224,4 +260,97 @@ mod tests {
             assert_eq!(part.len(), 4);
         }
     }
+
+    #[test]
+    fn test_json_to_ipld_cid_link() {
+        let json = serde_json::json!({
+            "$link": "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
+        });
+        let ipld = json_to_ipld(&json);
+        match ipld {
+            Ipld::Link(cid) => {
+                assert_eq!(
+                    cid.to_string(),
+                    "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
+                );
+            }
+            _ => panic!("Expected Ipld::Link, got {:?}", ipld),
+        }
+    }
+
+    #[test]
+    fn test_json_to_ipld_blob_ref() {
+        let json = serde_json::json!({
+            "$type": "blob",
+            "ref": {
+                "$link": "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
+            },
+            "mimeType": "image/jpeg",
+            "size": 12345
+        });
+        let ipld = json_to_ipld(&json);
+        match ipld {
+            Ipld::Map(map) => {
+                assert_eq!(map.get("$type"), Some(&Ipld::String("blob".to_string())));
+                assert_eq!(
+                    map.get("mimeType"),
+                    Some(&Ipld::String("image/jpeg".to_string()))
+                );
+                assert_eq!(map.get("size"), Some(&Ipld::Integer(12345)));
+                match map.get("ref") {
+                    Some(Ipld::Link(cid)) => {
+                        assert_eq!(
+                            cid.to_string(),
+                            "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
+                        );
+                    }
+                    _ => panic!("Expected Ipld::Link in ref field, got {:?}", map.get("ref")),
+                }
+            }
+            _ => panic!("Expected Ipld::Map, got {:?}", ipld),
+        }
+    }
+
+    #[test]
+    fn test_json_to_ipld_nested_blob_refs_serializes_correctly() {
+        let record = serde_json::json!({
+            "$type": "app.bsky.feed.post",
+            "text": "Hello world",
+            "embed": {
+                "$type": "app.bsky.embed.images",
+                "images": [
+                    {
+                        "alt": "Test image",
+                        "image": {
+                            "$type": "blob",
+                            "ref": {
+                                "$link": "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
+                            },
+                            "mimeType": "image/jpeg",
+                            "size": 12345
+                        }
+                    }
+                ]
+            }
+        });
+        let ipld = json_to_ipld(&record);
+        let cbor_bytes = serde_ipld_dagcbor::to_vec(&ipld).expect("CBOR serialization failed");
+        assert!(!cbor_bytes.is_empty());
+        let parsed: Ipld =
+            serde_ipld_dagcbor::from_slice(&cbor_bytes).expect("CBOR deserialization failed");
+        if let Ipld::Map(map) = &parsed
+            && let Some(Ipld::Map(embed)) = map.get("embed")
+            && let Some(Ipld::List(images)) = embed.get("images")
+            && let Some(Ipld::Map(img)) = images.first()
+            && let Some(Ipld::Map(blob)) = img.get("image")
+            && let Some(Ipld::Link(cid)) = blob.get("ref")
+        {
+            assert_eq!(
+                cid.to_string(),
+                "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
+            );
+            return;
+        }
+        panic!("Failed to find CID link in parsed CBOR");
+    }
 }