vtc-service 0.12.0

//! Encrypted full-state backup / restore (P3.9).
//!
//! The VTC holds a community's irreplaceable social state — members,
//! ACL, endorsements, relationships, policies, audit, and the bitstring
//! status lists whose loss bricks every issued VMC's `credentialStatus`.
//! This module exports that state (plus the signing key bundle) into a
//! single password-encrypted artifact and restores it.
//!
//! Design note: `docs/05-design-notes/vtc-backup-restore.md`. Crypto is
//! the VTA's verbatim — Argon2id + AES-256-GCM. The keyspace census
//! (`keyspaces::BACKED_UP` ∪ `EXCLUDED_FROM_BACKUP` == `ALL`) guarantees
//! no keyspace is silently omitted.

use std::collections::BTreeMap;

use aes_gcm::aead::Aead;
use aes_gcm::{Aes256Gcm, KeyInit, Nonce};
use argon2::Argon2;
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64;
use chrono::{DateTime, Utc};
use rand::Rng;
use serde::{Deserialize, Serialize};

use vti_common::error::AppError;

use crate::error::TaskError;

/// `vtc/backup/export:passwordTooShort`.
pub const EXPORT_ERR_PASSWORD_TOO_SHORT: &str =
    trust_tasks_rs::specs::vtc::backup::export::v0_1::error_codes::PASSWORD_TOO_SHORT.code;
/// `vtc/backup/import:decryptionFailed`.
pub const IMPORT_ERR_DECRYPTION_FAILED: &str =
    trust_tasks_rs::specs::vtc::backup::import::v0_1::error_codes::DECRYPTION_FAILED.code;
use vti_common::store::KeyspaceHandle;

use crate::config::MessagingConfig;
use crate::keys::seed_store::SecretStore;
use crate::server::AppState;
use crate::store::{Store, keyspaces};

// ── Crypto parameters (VTA-identical) ───────────────────────────────────
const VERSION: u32 = 1;
const FORMAT: &str = "vtc-backup-v1";
const ARGON2_M_COST: u32 = 65536; // 64 MiB
const ARGON2_T_COST: u32 = 3;
const ARGON2_P_COST: u32 = 4;
const SALT_LEN: usize = 32;
const NONCE_LEN: usize = 12;

// Import-side Argon2 bounds — clamp untrusted envelopes so a malicious
// `m_cost` can't drive a memory bomb on decrypt.
const MIN_M_COST: u32 = 8 * 1024; // 8 MiB
const MAX_M_COST: u32 = 1 << 20; // 1 GiB
const MIN_T_COST: u32 = 1;
const MAX_T_COST: u32 = 10;
const MIN_P_COST: u32 = 1;
const MAX_P_COST: u32 = 16;

/// Sentinel key, written into the `config` keyspace (excluded from
/// backup, so it survives the import-time clear) before the destructive
/// replay and removed only on success. Boot refuses to start while it's
/// present — see [`import_in_progress`] — **regardless of its age**.
///
/// The value is an RFC 3339 timestamp (when the in-flight import stamped
/// it), kept only so a refusal message or a recovery command can say
/// *when* the interrupted import started — it carries no expiry. The
/// sentinel means the backed-up keyspaces are in an indeterminate,
/// partially-cleared-and-replayed state (a half-written ACL or member
/// set is not "probably fine after a while"): only an explicit operator
/// decision can safely clear it. See [`discard_interrupted_import`] (`vtc
/// admin discard-interrupted-import`), the one thing that clears this
/// outside of [`import_backup`] completing.
const IMPORT_IN_PROGRESS_KEY: &[u8] = b"backup:import_in_progress";

// ── Wire types ──────────────────────────────────────────────────────────

/// Outer envelope: unencrypted metadata + the encrypted payload. Crypto
/// fields mirror the VTA's `vta-backup-v1`.
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
#[schema(as = VtcBackupEnvelope)]
#[serde(rename_all = "camelCase")]
pub struct BackupEnvelope {
    pub version: u32,
    pub format: String,
    pub created_at: DateTime<Utc>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub source_did: Option<String>,
    pub source_version: String,
    pub kdf: KdfParams,
    pub encryption: EncryptionParams,
    pub includes_audit: bool,
    /// base64url(AES-256-GCM(JSON([`BackupPayload`]))).
    pub ciphertext: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
#[schema(as = VtcBackupKdfParams)]
#[serde(rename_all = "camelCase")]
pub struct KdfParams {
    pub algorithm: String, // "argon2id"
    pub salt: String,      // base64url, 32 bytes
    pub m_cost: u32,
    pub t_cost: u32,
    pub p_cost: u32,
}

#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
#[schema(as = VtcBackupEncryptionParams)]
#[serde(rename_all = "camelCase")]
pub struct EncryptionParams {
    pub algorithm: String, // "aes-256-gcm"
    pub nonce: String,     // base64url, 12 bytes
}

/// Inner (encrypted) payload. A config snapshot, the signing key bundle,
/// and a faithful raw dump of every backed-up keyspace.
#[derive(Clone, Serialize, Deserialize)]
pub struct BackupPayload {
    pub config: BackupConfig,
    /// Hex of the raw secret-store bytes (the encoded `VtcKeyBundle`).
    /// `None` only if the source had no bundle stored.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_bundle_hex: Option<String>,
    pub keyspaces: Vec<KeyspaceDump>,
}

/// Written by hand so the signing key bundle never reaches a log: a derived `Debug` would print it.
impl std::fmt::Debug for BackupPayload {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BackupPayload")
            .field("config", &self.config)
            .field(
                "key_bundle_hex",
                &self.key_bundle_hex.as_ref().map(|_| "<redacted>"),
            )
            .field("keyspaces", &self.keyspaces)
            .finish()
    }
}

/// One backed-up keyspace's full contents. Both key and value are
/// base64url-encoded so binary keys/values round-trip losslessly.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KeyspaceDump {
    pub name: String,
    pub rows: Vec<(String, String)>,
}

/// The slice of config that travels with a backup so a restore onto a
/// fresh install reconstitutes the VTC's identity. The secrets *backend*
/// is deliberately not carried — the target's own backend is used.
#[derive(Clone, Default, Serialize, Deserialize)]
pub struct BackupConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub vtc_did: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub vtc_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub vta_did: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub public_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub messaging: Option<MessagingConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub jwt_signing_key: Option<String>,
}

/// Written by hand so the JWT signing key never reaches a log: a derived `Debug` would print it.
impl std::fmt::Debug for BackupConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BackupConfig")
            .field("vtc_did", &self.vtc_did)
            .field("vtc_name", &self.vtc_name)
            .field("vta_did", &self.vta_did)
            .field("public_url", &self.public_url)
            .field("messaging", &self.messaging)
            .field(
                "jwt_signing_key",
                &self.jwt_signing_key.as_ref().map(|_| "<redacted>"),
            )
            .finish()
    }
}

/// Result of an import (or, with `confirm = false`, a preview).
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
#[schema(as = VtcBackupImportResult)]
#[serde(rename_all = "camelCase")]
pub struct ImportResult {
    pub status: String, // "imported" | "preview"
    #[serde(skip_serializing_if = "Option::is_none")]
    pub source_did: Option<String>,
    /// Per-keyspace row counts the import wrote (or would write).
    pub counts: BTreeMap<String, usize>,
    pub message: String,
}

// ── Export ──────────────────────────────────────────────────────────────

/// Export every backed-up keyspace + the signing key bundle into an
/// encrypted [`BackupEnvelope`]. The `secret_store` (the configured
/// backend, constructed by the route from config) supplies the signing
/// key bundle. Caller must be super-admin (gated at the route).
pub async fn export_backup(
    state: &AppState,
    secret_store: &dyn SecretStore,
    password: &str,
    include_audit: bool,
) -> Result<BackupEnvelope, TaskError> {
    // `vtc/backup/export:passwordTooShort`. The minimum is the workspace's
    // `MIN_BACKUP_PASSWORD_LEN` (15), stricter than the 12 the specification
    // names; a stricter floor is not lowered to match (#1600).
    vta_sdk::protocols::backup_management::validate_backup_password(password)
        .map_err(|e| TaskError::declared(EXPORT_ERR_PASSWORD_TOO_SHORT, AppError::Validation(e)))?;

    // Signing key bundle (hex of the raw stored bytes — backend-agnostic
    // round-trip).
    let key_bundle_hex = secret_store.get().await?.map(hex::encode);

    // Config snapshot.
    let config = {
        let cfg = state.config.read().await;
        BackupConfig {
            vtc_did: cfg.vtc_did.clone(),
            vtc_name: cfg.vtc_name.clone(),
            vta_did: cfg.vta_did.clone(),
            public_url: cfg.public_url.clone(),
            messaging: cfg.messaging.clone(),
            jwt_signing_key: cfg.auth.jwt_signing_key.clone(),
        }
    };

    // Keyspace census. `audit` rows are skipped unless requested; the
    // keyspace still appears (empty) so the dump shape is stable.
    //
    // `audit_checkpoint` is gated on the *same* flag, and must stay that way.
    // The two only mean anything together: checkpoints restored without their
    // log attest to entries that aren't there (reads as total truncation), and
    // a log restored without its checkpoints is shorter than every signed
    // checkpoint claims (also reads as truncation). Either half alone turns a
    // legitimate restore into the exact finding #708 exists to raise.
    let mut keyspace_dumps = Vec::with_capacity(keyspaces::BACKED_UP.len());
    for name in keyspaces::BACKED_UP {
        if (*name == keyspaces::AUDIT || *name == keyspaces::AUDIT_CHECKPOINT) && !include_audit {
            keyspace_dumps.push(KeyspaceDump {
                name: (*name).to_string(),
                rows: Vec::new(),
            });
            continue;
        }
        let ks = backed_up_handle(state, name).ok_or_else(|| {
            AppError::Internal(format!("backup: no AppState handle for keyspace '{name}'"))
        })?;
        let raw = ks.prefix_iter_raw(Vec::<u8>::new()).await?;
        let rows = raw
            .into_iter()
            .map(|(k, v)| (BASE64.encode(k), BASE64.encode(v)))
            .collect();
        keyspace_dumps.push(KeyspaceDump {
            name: (*name).to_string(),
            rows,
        });
    }

    let payload = BackupPayload {
        config,
        key_bundle_hex,
        keyspaces: keyspace_dumps,
    };
    Ok(encrypt_payload(&payload, password, include_audit, state).await?)
}

// ── Import ──────────────────────────────────────────────────────────────

/// Decrypt + apply a backup. With `confirm = false` this is a no-op
/// **preview** returning the row counts; with `confirm = true` it clears
/// the backed-up keyspaces and replays the backup. Identity-guarded on
/// `vtc_did`. Caller must be super-admin (gated at the route).
pub async fn import_backup(
    state: &AppState,
    secret_store: &dyn SecretStore,
    envelope: &BackupEnvelope,
    password: &str,
    confirm: bool,
) -> Result<ImportResult, TaskError> {
    // `vtc/backup/import:decryptionFailed` — the GCM tag did not verify, so
    // the password is wrong or the ciphertext was altered. The envelope's
    // shape checks (format, KDF bounds, encodings) stay undeclared 400s.
    let payload = decrypt_backup(envelope, password).map_err(|e| match e {
        e @ AppError::Authentication(_) => TaskError::declared(IMPORT_ERR_DECRYPTION_FAILED, e),
        e => TaskError::App(e),
    })?;

    // Identity guard before any mutation.
    {
        let running = state.config.read().await.vtc_did.clone();
        check_vtc_did_compatibility(running.as_deref(), payload.config.vtc_did.as_deref())?;
    }

    let counts: BTreeMap<String, usize> = payload
        .keyspaces
        .iter()
        .map(|d| (d.name.clone(), d.rows.len()))
        .collect();

    // ACL rows written before role-based administration are mapped onto it
    // here (`vtc-admin-roles.md` §9) — decided before anything is written, so
    // a row that cannot be mapped refuses the import rather than half-applying
    // it, and the preview reports the same re-grants the import will.
    let acl_rows = migrate_acl_rows(&payload)?;
    let regrant = regrant_note(&acl_rows);

    if !confirm {
        return Ok(ImportResult {
            status: "preview".into(),
            source_did: payload.config.vtc_did.clone(),
            counts,
            message: format!(
                "Preview only — pass confirm=true to apply. This will overwrite all community \
                 state.{regrant}"
            ),
        });
    }

    // Crash-safety: stamp the sentinel (in the excluded `config`
    // keyspace, so it survives the clear) + flush before mutating. The
    // value is *when*, not just *that* — purely informational (a refusal
    // message or `discard-interrupted-import` names it); it carries no
    // expiry, see [`IMPORT_IN_PROGRESS_KEY`].
    state
        .config_ks
        .insert_raw(
            IMPORT_IN_PROGRESS_KEY.to_vec(),
            Utc::now().to_rfc3339().into_bytes(),
        )
        .await?;
    state.config_ks.persist().await?;

    // Clear every backed-up keyspace, then replay.
    for name in keyspaces::BACKED_UP {
        let ks = backed_up_handle(state, name).ok_or_else(|| {
            AppError::Internal(format!("backup: no AppState handle for keyspace '{name}'"))
        })?;
        clear_keyspace(ks).await?;
    }
    for dump in &payload.keyspaces {
        // Ignore a keyspace the dump names but we don't back up (forward
        // compat with a future BACKED_UP addition).
        let Some(ks) = backed_up_handle(state, &dump.name) else {
            continue;
        };
        for (k_b64, v_b64) in &dump.rows {
            let key = BASE64
                .decode(k_b64)
                .map_err(|e| AppError::Validation(format!("backup: bad row key b64: {e}")))?;
            let val = BASE64
                .decode(v_b64)
                .map_err(|e| AppError::Validation(format!("backup: bad row value b64: {e}")))?;
            // An ACL row is written as migrated, never as it arrived.
            let val = match acl_rows.iter().find(|(k, _, _)| *k == key) {
                Some((_, entry, _)) => serde_json::to_vec(entry)
                    .map_err(|e| AppError::Internal(format!("backup: encode ACL row: {e}")))?,
                None => val,
            };
            ks.insert_raw(key, val).await?;
        }
        ks.persist().await?;
    }

    // Restore identity config + the signing key bundle.
    apply_config(
        state,
        secret_store,
        &payload.config,
        payload.key_bundle_hex.as_deref(),
    )
    .await?;

    // Done — clear the sentinel + flush.
    state
        .config_ks
        .remove(IMPORT_IN_PROGRESS_KEY.to_vec())
        .await?;
    state.config_ks.persist().await?;

    // The restored configs, not the ones this process opened with.
    if let Err(e) = state.room_storage.reload_stores().await {
        tracing::warn!(error = %e, "backup import: storage configs not reloaded; the restart reloads them");
    }

    // A backup taken before phase C3 carries its git rights in the old rights
    // store: move them onto the restored ACL entries now (`vtc-admin-roles.md`
    // §9). Boot runs the same migration, so a failure here is logged and
    // finished by the restart the import asks for.
    if let Err(e) = crate::acl::resource_grant::rebuild_holder_index(&state.acl_ks).await {
        tracing::warn!(error = %e, "backup import: the git grant-holder index could not be rebuilt; boot rebuilds it");
    }
    if let Err(e) = crate::git_ns::migrate::migrate_rights(state, "backup import").await {
        tracing::warn!(error = %e, "backup import: git-namespace rights not yet moved onto the ACL entries; boot finishes it");
    }

    Ok(ImportResult {
        status: "imported".into(),
        source_did: payload.config.vtc_did.clone(),
        counts,
        message: format!(
            "Import complete. Restart the daemon to serve the restored identity.{regrant}"
        ),
    })
}

/// Every `acl:` row of the backup's ACL keyspace, mapped onto the role-based
/// shape: `(raw key, entry, what it became)`.
fn migrate_acl_rows(
    payload: &BackupPayload,
) -> Result<
    Vec<(
        Vec<u8>,
        crate::acl::VtcAclEntry,
        crate::acl::migrate::Migrated,
    )>,
    AppError,
> {
    let mut out = Vec::new();
    for dump in payload
        .keyspaces
        .iter()
        .filter(|d| d.name == keyspaces::ACL)
    {
        for (k_b64, v_b64) in &dump.rows {
            let key = BASE64
                .decode(k_b64)
                .map_err(|e| AppError::Validation(format!("backup: bad row key b64: {e}")))?;
            if !key.starts_with(b"acl:") {
                continue;
            }
            let val = BASE64
                .decode(v_b64)
                .map_err(|e| AppError::Validation(format!("backup: bad row value b64: {e}")))?;
            let (entry, how) = crate::acl::migrate::migrate_row(&val)?;
            out.push((key, entry, how));
        }
    }
    Ok(out)
}

/// The import report's re-grant list: context-scoped admins that came across
/// with no administrative role (`vtc-admin-roles.md` §9).
fn regrant_note(
    rows: &[(
        Vec<u8>,
        crate::acl::VtcAclEntry,
        crate::acl::migrate::Migrated,
    )],
) -> String {
    let dids: Vec<&str> = rows
        .iter()
        .filter(|(_, _, how)| matches!(how, crate::acl::migrate::Migrated::NeedsRegrant { .. }))
        .map(|(_, e, _)| e.did.as_str())
        .collect();
    if dids.is_empty() {
        String::new()
    } else {
        format!(
            " {} context-scoped administrator entr{} came across with no administrative role \
             and need re-granting with acl/update: {}",
            dids.len(),
            if dids.len() == 1 { "y" } else { "ies" },
            dids.join(", ")
        )
    }
}

/// True if a previous import was interrupted before it finished. Boot
/// consults this and refuses to start while it's set — **whatever its
/// age** — because the half-applied state (a partially cleared, partially
/// replayed keyspace set) is unsafe to serve. There is no TTL: a stale
/// marker does not mean the state became any safer to serve, only that
/// nobody has looked at it yet. Cleared by a successful re-import, or by
/// the deliberate [`discard_interrupted_import`] recovery path.
pub async fn import_in_progress(config_ks: &KeyspaceHandle) -> Result<bool, AppError> {
    Ok(import_in_progress_since(config_ks).await?.is_some())
}

/// When the interrupted import stamped the sentinel, as the raw RFC 3339
/// string it was written with (not reparsed — this is for a human to
/// read in a refusal message or a recovery command's confirmation
/// prompt, never for an age comparison). `None` if no import is in
/// progress.
pub async fn import_in_progress_since(
    config_ks: &KeyspaceHandle,
) -> Result<Option<String>, AppError> {
    let stamp = config_ks
        .prefix_iter_raw(IMPORT_IN_PROGRESS_KEY.to_vec())
        .await?
        .into_iter()
        .find(|(k, _)| k.as_slice() == IMPORT_IN_PROGRESS_KEY)
        .map(|(_, v)| v);
    Ok(stamp.map(|v| {
        String::from_utf8(v)
            .unwrap_or_else(|e| format!("<unreadable stamp: {} bytes>", e.into_bytes().len()))
    }))
}

/// Explicit operator recovery from an import a crash, a client disconnect,
/// or a `finalize-import` that never landed left interrupted: wipe every
/// backed-up keyspace back to empty and clear the sentinel. Fronted by
/// `vtc admin discard-interrupted-import` — never invoked automatically,
/// and never implied by the sentinel's age.
///
/// This is the "start over from empty" recovery, not a restore: it does
/// not re-apply the original backup, because that bundle isn't
/// necessarily still at hand (and this runs offline, with no password
/// prompt). An operator who still has the bundle should prefer re-running
/// the normal import with it once this clears the sentinel — that
/// restores real state instead of leaving the community empty.
///
/// Returns the interrupted import's stamp (as read, unparsed) if a
/// sentinel was present and has now been cleared; `None` if there was
/// nothing to discard (left entirely untouched in that case).
pub async fn discard_interrupted_import(store: &Store) -> Result<Option<String>, AppError> {
    let config_ks = store.keyspace(keyspaces::CONFIG)?;
    let Some(since) = import_in_progress_since(&config_ks).await? else {
        return Ok(None);
    };

    for name in keyspaces::BACKED_UP {
        let ks = store.keyspace(name)?;
        clear_keyspace(&ks).await?;
    }

    config_ks.remove(IMPORT_IN_PROGRESS_KEY.to_vec()).await?;
    config_ks.persist().await?;
    Ok(Some(since))
}

/// Test-only: stamp the import-in-progress sentinel as if an import had
/// started `at`, without actually running one — so a test can simulate a
/// crash mid-import (stamped, never cleared) and exercise the boot
/// refusal / recovery path without a real one. Never compiled into a
/// production build.
#[cfg(feature = "test-support")]
pub async fn stamp_import_in_progress_for_test(
    config_ks: &KeyspaceHandle,
    at: DateTime<Utc>,
) -> Result<(), AppError> {
    config_ks
        .insert_raw(
            IMPORT_IN_PROGRESS_KEY.to_vec(),
            at.to_rfc3339().into_bytes(),
        )
        .await?;
    config_ks.persist().await?;
    Ok(())
}

/// `vtc_did` guard — a configured VTC refuses a backup from a different
/// identity; a fresh install accepts anything. Mirrors the VTA's
/// `check_vta_did_compatibility`.
fn check_vtc_did_compatibility(
    running_did: Option<&str>,
    backup_did: Option<&str>,
) -> Result<(), AppError> {
    let running = match running_did {
        Some(d) if !d.is_empty() => d,
        _ => return Ok(()), // fresh install accepts any backup
    };
    let backup = backup_did.unwrap_or("");
    if backup == running {
        return Ok(());
    }
    Err(AppError::Conflict(format!(
        "backup vtc_did mismatch: backup claims '{backup}' but this VTC is running as \
         '{running}'. Refusing to overwrite identity. If this is intentional (identity \
         migration), clear vtc_did from the running config first."
    )))
}

// ── Internals ───────────────────────────────────────────────────────────

/// Map a backed-up keyspace name to its `AppState` handle. Returns
/// `None` for names not in `BACKED_UP`.
fn backed_up_handle<'a>(state: &'a AppState, name: &str) -> Option<&'a KeyspaceHandle> {
    use keyspaces::*;
    Some(match name {
        x if x == ACL => &state.acl_ks,
        x if x == COMMUNITY => &state.community_ks,
        x if x == MEMBERS => &state.members_ks,
        x if x == JOIN_REQUESTS => &state.join_requests_ks,
        x if x == POLICIES => &state.policies_ks,
        x if x == ACTIVE_POLICIES => &state.active_policies_ks,
        x if x == STATUS_LISTS => &state.status_lists_ks,
        x if x == RELATIONSHIPS => &state.relationships_ks,
        x if x == RELATIONSHIPS_BY_DID => &state.relationships_by_did_ks,
        x if x == ENDORSEMENT_TYPES => &state.endorsement_types_ks,
        x if x == SCHEMAS => &state.schemas_ks,
        x if x == ENDORSEMENTS => &state.endorsements_ks,
        x if x == VETTING_REVOCATIONS => &state.vetting_revocations_ks,
        x if x == VETTER_PROFILES => &state.vetter_profiles_ks,
        x if x == VETTING_PCS_SPENT => &state.vetting_pcs_spent_ks,
        x if x == VETTING_PCS_ISSUE => &state.vetting_pcs_issue_ks,
        // A room's records are backed up like any other community state. On a
        // sealed tier they are ciphertext this service cannot read, and that is
        // no reason to skip them: the host is trusted for availability
        // (invariant I2), and losing the ciphertext loses the room just as
        // completely as losing plaintext would.
        x if x == ROOMS => &state.rooms_ks,
        x if x == ROOM_RECORDS => &state.room_records_ks,
        x if x == ROOM_EPOCH_LINKS => &state.room_epoch_links_ks,
        x if x == ROOM_REVOCATIONS => &state.room_revocations_ks,
        x if x == ROOM_BLOBS => state.room_files.index_keyspace(),
        x if x == STORAGE_CONFIGS => state.room_storage.keyspace(),
        x if x == COMMUNITY_ROOMS => state.community_rooms.keyspace(),
        x if x == DIRECTORY => state.directory.ks(),
        x if x == INVITATIONS => &state.invitations_ks,
        x if x == CONSUMED_INVITATIONS => &state.consumed_invitations_ks,
        x if x == AUDIT => &state.audit_ks,
        x if x == AUDIT_KEY => &state.audit_key_ks,
        x if x == AUDIT_CHECKPOINT => &state.audit_checkpoint_ks,
        // The git-namespace records are the source of truth for every right
        // this community has published about its repositories; the registry
        // and the forge are rebuilt from them.
        x if x == GIT_NS => &state.git_ns.ks,
        x if x == STEP_UP_APPROVERS => &state.step_up_approvers_ks,
        // The encrypted handle, like `audit_key`: export reads the plaintext
        // rows, and an import writes them back under the storage key the
        // restored key bundle reproduces.
        x if x == OID4VP_KEYS => &state.oid4vp_keys_ks,
        _ => return None,
    })
}

/// Remove every row from a keyspace (whole-keyspace clear — VTC
/// keyspaces are single-purpose, unlike the VTA's prefix-shared ones).
async fn clear_keyspace(ks: &KeyspaceHandle) -> Result<(), AppError> {
    let keys: Vec<Vec<u8>> = ks
        .prefix_iter_raw(Vec::<u8>::new())
        .await?
        .into_iter()
        .map(|(k, _)| k)
        .collect();
    for k in keys {
        ks.remove(k).await?;
    }
    Ok(())
}

/// Apply the config snapshot + restore the signing key bundle, then
/// persist `config.toml`.
async fn apply_config(
    state: &AppState,
    secret_store: &dyn SecretStore,
    bc: &BackupConfig,
    key_bundle_hex: Option<&str>,
) -> Result<(), AppError> {
    // Restore the bundle first. The config-secret backend is read-only
    // at runtime (its bytes live inline in config.toml), so for it we
    // write the hex into the config; every other backend takes a `set`.
    if let Some(hex_s) = key_bundle_hex {
        let bundle_bytes = hex::decode(hex_s)
            .map_err(|e| AppError::Validation(format!("backup: bad key_bundle hex: {e}")))?;
        let is_inline = state.config.read().await.secrets.secret.is_some();
        if is_inline {
            state.config.write().await.secrets.secret = Some(hex::encode(&bundle_bytes));
        } else {
            secret_store
                .set(&bundle_bytes)
                .await
                .map_err(|e| AppError::SecretStore(format!("backup: restore key bundle: {e}")))?;
        }
    }

    let mut cfg = state.config.write().await;
    if let Some(v) = &bc.vtc_did {
        cfg.vtc_did = Some(v.clone());
    }
    if let Some(v) = &bc.vtc_name {
        cfg.vtc_name = Some(v.clone());
    }
    if let Some(v) = &bc.vta_did {
        cfg.vta_did = Some(v.clone());
    }
    if let Some(v) = &bc.public_url {
        cfg.public_url = Some(v.clone());
    }
    if bc.messaging.is_some() {
        cfg.messaging = bc.messaging.clone();
    }
    if let Some(v) = &bc.jwt_signing_key {
        cfg.auth.jwt_signing_key = Some(v.clone());
    }
    cfg.save()?;
    Ok(())
}

fn encrypt_payload_inner(
    payload: &BackupPayload,
    password: &str,
    include_audit: bool,
    source_did: Option<String>,
) -> Result<BackupEnvelope, AppError> {
    let plaintext = serde_json::to_vec(payload)
        .map_err(|e| AppError::Internal(format!("backup serialize: {e}")))?;

    let mut salt = [0u8; SALT_LEN];
    rand::rng().fill_bytes(&mut salt);
    let mut nonce_bytes = [0u8; NONCE_LEN];
    rand::rng().fill_bytes(&mut nonce_bytes);

    let key = derive_key(password, &salt, ARGON2_M_COST, ARGON2_T_COST, ARGON2_P_COST)?;
    let cipher = Aes256Gcm::new_from_slice(&key)
        .map_err(|e| AppError::Internal(format!("backup aes key: {e}")))?;
    let ciphertext = cipher
        .encrypt((&nonce_bytes).into(), plaintext.as_ref())
        .map_err(|e| AppError::Internal(format!("backup encrypt: {e}")))?;

    Ok(BackupEnvelope {
        version: VERSION,
        format: FORMAT.into(),
        created_at: Utc::now(),
        source_did,
        source_version: env!("CARGO_PKG_VERSION").into(),
        kdf: KdfParams {
            algorithm: "argon2id".into(),
            salt: BASE64.encode(salt),
            m_cost: ARGON2_M_COST,
            t_cost: ARGON2_T_COST,
            p_cost: ARGON2_P_COST,
        },
        encryption: EncryptionParams {
            algorithm: "aes-256-gcm".into(),
            nonce: BASE64.encode(nonce_bytes),
        },
        includes_audit: include_audit,
        ciphertext: BASE64.encode(&ciphertext),
    })
}

async fn encrypt_payload(
    payload: &BackupPayload,
    password: &str,
    include_audit: bool,
    state: &AppState,
) -> Result<BackupEnvelope, AppError> {
    let source_did = state.config.read().await.vtc_did.clone();
    encrypt_payload_inner(payload, password, include_audit, source_did)
}

/// Decrypt + validate a backup envelope. Wrong password / tamper →
/// `Authentication` (the GCM tag failed).
pub fn decrypt_backup(
    envelope: &BackupEnvelope,
    password: &str,
) -> Result<BackupPayload, AppError> {
    if envelope.version != VERSION || envelope.format != FORMAT {
        return Err(AppError::Validation(format!(
            "unsupported backup format: {} v{}",
            envelope.format, envelope.version
        )));
    }
    if envelope.kdf.algorithm != "argon2id" {
        return Err(AppError::Validation(format!(
            "unsupported KDF '{}' (only argon2id)",
            envelope.kdf.algorithm
        )));
    }
    if !(MIN_M_COST..=MAX_M_COST).contains(&envelope.kdf.m_cost) {
        return Err(AppError::Validation(format!(
            "argon2 m_cost {} out of bounds [{MIN_M_COST}, {MAX_M_COST}]",
            envelope.kdf.m_cost
        )));
    }
    if !(MIN_T_COST..=MAX_T_COST).contains(&envelope.kdf.t_cost) {
        return Err(AppError::Validation("argon2 t_cost out of bounds".into()));
    }
    if !(MIN_P_COST..=MAX_P_COST).contains(&envelope.kdf.p_cost) {
        return Err(AppError::Validation("argon2 p_cost out of bounds".into()));
    }
    if envelope.encryption.algorithm != "aes-256-gcm" {
        return Err(AppError::Validation(format!(
            "unsupported cipher '{}' (only aes-256-gcm)",
            envelope.encryption.algorithm
        )));
    }

    let salt = BASE64
        .decode(&envelope.kdf.salt)
        .map_err(|e| AppError::Validation(format!("invalid salt: {e}")))?;
    if salt.len() != SALT_LEN {
        return Err(AppError::Validation(format!(
            "invalid salt length {} (expected {SALT_LEN})",
            salt.len()
        )));
    }
    let nonce_bytes = BASE64
        .decode(&envelope.encryption.nonce)
        .map_err(|e| AppError::Validation(format!("invalid nonce: {e}")))?;
    if nonce_bytes.len() != NONCE_LEN {
        return Err(AppError::Validation(format!(
            "invalid nonce length {} (expected {NONCE_LEN})",
            nonce_bytes.len()
        )));
    }
    let ciphertext = BASE64
        .decode(&envelope.ciphertext)
        .map_err(|e| AppError::Validation(format!("invalid ciphertext: {e}")))?;

    let key = derive_key(
        password,
        &salt,
        envelope.kdf.m_cost,
        envelope.kdf.t_cost,
        envelope.kdf.p_cost,
    )?;
    let cipher = Aes256Gcm::new_from_slice(&key)
        .map_err(|e| AppError::Internal(format!("backup aes key: {e}")))?;
    // `try_from`, not the deprecated `from_slice`, which panicked on a wrong
    // length. The check at the top of this function already rejects that, so
    // this cannot fail — the point is that it can no longer panic if that
    // check is ever moved or removed.
    let nonce = Nonce::try_from(nonce_bytes.as_slice())
        .map_err(|_| AppError::Validation(format!("nonce must be {NONCE_LEN} bytes")))?;
    let plaintext = cipher
        .decrypt(&nonce, ciphertext.as_ref())
        .map_err(|_| AppError::Authentication("incorrect backup password".into()))?;

    serde_json::from_slice(&plaintext)
        .map_err(|e| AppError::Internal(format!("backup payload corrupt: {e}")))
}

fn derive_key(
    password: &str,
    salt: &[u8],
    m_cost: u32,
    t_cost: u32,
    p_cost: u32,
) -> Result<[u8; 32], AppError> {
    let argon2 = Argon2::new(
        argon2::Algorithm::Argon2id,
        argon2::Version::V0x13,
        argon2::Params::new(m_cost, t_cost, p_cost, Some(32))
            .map_err(|e| AppError::Validation(format!("argon2 params: {e}")))?,
    );
    let mut key = [0u8; 32];
    argon2
        .hash_password_into(password.as_bytes(), salt, &mut key)
        .map_err(|e| AppError::Internal(format!("argon2 hash: {e}")))?;
    Ok(key)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// `BACKED_UP` and `backed_up_handle` are two lists of the same thing, and nothing in
    /// the type system makes them agree — the match ends in `_ => return None`, so a
    /// keyspace added to one and not the other fails at *backup time* with
    /// "no AppState handle for keyspace", which is a runtime error on the one operation an
    /// operator runs when something has already gone wrong.
    ///
    /// That is exactly how `rooms` and `room_records` broke every backup test: added to the
    /// partition, absent from the map, and no compile error anywhere. This pins the two
    /// together the way the partition census pins the partition.
    #[tokio::test]
    async fn every_backed_up_keyspace_has_a_handle() {
        let tv = crate::test_support::build_test_vtc().await;
        let missing: Vec<&str> = keyspaces::BACKED_UP
            .iter()
            .filter(|name| backed_up_handle(&tv.state, name).is_none())
            .copied()
            .collect();
        assert!(
            missing.is_empty(),
            "these keyspaces are in BACKED_UP but `backed_up_handle` does not map them, so \
             every export would fail at runtime: {missing:?}"
        );
    }

    /// The other direction: a handle for a keyspace nothing backs up is dead code that
    /// reads as coverage.
    #[tokio::test]
    async fn no_handle_maps_a_keyspace_that_is_not_backed_up() {
        let tv = crate::test_support::build_test_vtc().await;
        let stray: Vec<&str> = keyspaces::EXCLUDED_FROM_BACKUP
            .iter()
            .filter(|name| backed_up_handle(&tv.state, name).is_some())
            .copied()
            .collect();
        assert!(
            stray.is_empty(),
            "`backed_up_handle` maps keyspaces excluded from backup: {stray:?}"
        );
    }

    fn sample_payload() -> BackupPayload {
        BackupPayload {
            config: BackupConfig {
                vtc_did: Some("did:webvh:vtc.example.com:abc".into()),
                vtc_name: Some("Acme".into()),
                vta_did: Some("did:webvh:vta.example.com:xyz".into()),
                public_url: Some("https://vtc.example.com".into()),
                messaging: None,
                jwt_signing_key: Some("sign-key".into()),
            },
            key_bundle_hex: Some(hex::encode(b"bundle-bytes")),
            keyspaces: vec![
                KeyspaceDump {
                    name: keyspaces::ACL.into(),
                    rows: vec![(
                        BASE64.encode("acl:did:key:z6Mk"),
                        BASE64.encode(r#"{"r":1}"#),
                    )],
                },
                KeyspaceDump {
                    name: keyspaces::STATUS_LISTS.into(),
                    // a binary-ish value to prove base64 round-trips it
                    rows: vec![(BASE64.encode("sl:0"), BASE64.encode([0u8, 159, 146, 150]))],
                },
            ],
        }
    }

    fn enc(p: &BackupPayload, pw: &str) -> BackupEnvelope {
        encrypt_payload_inner(p, pw, false, p.config.vtc_did.clone()).unwrap()
    }

    #[test]
    fn roundtrip() {
        let p = sample_payload();
        let env = enc(&p, "correct-horse-battery");
        assert_eq!(env.version, 1);
        assert_eq!(env.format, "vtc-backup-v1");
        assert_eq!(env.kdf.algorithm, "argon2id");
        assert_eq!(env.encryption.algorithm, "aes-256-gcm");

        let back = decrypt_backup(&env, "correct-horse-battery").unwrap();
        assert_eq!(back.config.vtc_did, p.config.vtc_did);
        assert_eq!(back.key_bundle_hex, p.key_bundle_hex);
        assert_eq!(back.keyspaces.len(), 2);
        assert_eq!(
            back.keyspaces[1].rows[0].1,
            BASE64.encode([0u8, 159, 146, 150])
        );
    }

    #[test]
    fn wrong_password_fails() {
        let env = enc(&sample_payload(), "the-right-password");
        let err = decrypt_backup(&env, "the-wrong-password").unwrap_err();
        assert!(format!("{err}").contains("incorrect backup password"));
    }

    #[test]
    fn tampered_ciphertext_detected() {
        let mut env = enc(&sample_payload(), "twelve-char-pw!!");
        let mut ct = BASE64.decode(&env.ciphertext).unwrap();
        *ct.last_mut().unwrap() ^= 0xFF;
        env.ciphertext = BASE64.encode(&ct);
        assert!(decrypt_backup(&env, "twelve-char-pw!!").is_err());
    }

    #[test]
    fn rejects_bad_format_and_param_bounds() {
        let good = enc(&sample_payload(), "twelve-char-pw!!");

        let mut bad = good.clone();
        bad.format = "vta-backup-v1".into();
        assert!(decrypt_backup(&bad, "twelve-char-pw!!").is_err());

        let mut bad = good.clone();
        bad.kdf.m_cost = MAX_M_COST + 1;
        assert!(decrypt_backup(&bad, "twelve-char-pw!!").is_err());

        let mut bad = good.clone();
        bad.kdf.algorithm = "scrypt".into();
        assert!(decrypt_backup(&bad, "twelve-char-pw!!").is_err());

        let mut bad = good;
        bad.kdf.salt = BASE64.encode([0u8; 8]); // wrong length
        assert!(decrypt_backup(&bad, "twelve-char-pw!!").is_err());
    }

    #[test]
    fn did_guard_fresh_install_accepts_any() {
        check_vtc_did_compatibility(None, Some("did:key:z6MkAny")).unwrap();
        check_vtc_did_compatibility(Some(""), Some("did:key:z6MkAny")).unwrap();
        check_vtc_did_compatibility(None, None).unwrap();
    }

    #[test]
    fn did_guard_matching_accepted() {
        check_vtc_did_compatibility(Some("did:key:z6MkSame"), Some("did:key:z6MkSame")).unwrap();
    }

    #[test]
    fn did_guard_mismatch_rejected() {
        let err =
            check_vtc_did_compatibility(Some("did:key:z6MkRunning"), Some("did:key:z6MkForeign"))
                .unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("vtc_did mismatch"), "{msg}");
        assert!(
            msg.contains("z6MkForeign") && msg.contains("z6MkRunning"),
            "{msg}"
        );
    }
}