arcly-http 0.4.0

//! Field-level envelope encryption + crypto-shredding.
//!
//! ## Why this is load-bearing
//!
//! The durability machinery (outbox rows, idempotency replay cache, DLQ,
//! hash-chained audit) copies payloads into stores that are append-only *on
//! purpose*. Masking protects what leaves over HTTP; this module protects
//! what stays. Sensitive fields are sealed with a per-tenant or per-subject
//! data key (DEK) before they reach any sink, and GDPR Art. 17 erasure is
//! satisfied by destroying the DEK (**crypto-shredding**) — every ciphertext
//! copy, including ones inside the sealed audit chain, becomes permanently
//! unreadable without breaking the chain's integrity.
//!
//! ## Key hierarchy
//!
//! - **KEK** (key-encryption-key) lives in the [`KekSource`] — Vault / cloud
//!   KMS in production, an env-derived source in dev. The framework never
//!   sees it; the source hands back *unwrapped DEKs*.
//! - **DEK** (data-encryption-key, AES-256) is scoped by [`KeyId`]:
//!   `tenant:acme` for tenant-wide data, `subject:user-42` for per-person
//!   shredding. Rotation adds a new DEK *version*; old versions stay
//!   readable until re-encryption, because every ciphertext records the
//!   version that sealed it.
//!
//! ## Zero-lock mechanics
//!
//! The unwrapped key ring lives behind one `ArcSwap` snapshot — the proven
//! pattern from secrets / tenants / masking. `encrypt`/`decrypt` cost one
//! atomic pointer load, one hash probe, and one AES-256-GCM operation
//! (AES-NI). All I/O — provisioning, rotation, shredding — happens on the
//! control plane, which serializes through a tokio mutex that no request
//! path ever touches.
//!
//! ## Usage
//!
//! ```ignore
//! // boot (plugin on_init):
//! let vault = CryptoVault::bootstrap(Arc::new(VaultKekSource::new(...))).await?;
//! ctx.provide(vault);
//!
//! // declare what to seal on the DTO:
//! #[EncryptFields(key = "tenant:acme", fields("ssn", "card.number"))]
//! #[derive(serde::Serialize, serde::Deserialize)]
//! struct PatientRecord { ssn: String, card: Card, name: String }
//!
//! // write path — seal before any sink:
//! let sealed: serde_json::Value = record.seal(vault)?;
//!
//! // read path — unseal after load:
//! let record = PatientRecord::unseal(sealed, vault)?;
//!
//! // GDPR erasure — every copy of this subject's data dies at once:
//! vault.shred(&KeyId::subject("user-42")).await?;
//! ```

use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;

use aes_gcm::aead::{Aead, KeyInit, OsRng};
use aes_gcm::{AeadCore, Aes256Gcm, Key, Nonce};
use arc_swap::ArcSwap;
use base64::Engine;
use futures::future::BoxFuture;
use secrecy::{ExposeSecret, Secret};
use serde_json::Value;
use smol_str::SmolStr;

const WIRE_PREFIX: &str = "enc:v1:";
const B64: base64::engine::GeneralPurpose = base64::engine::general_purpose::STANDARD_NO_PAD;

// ─── Errors ───────────────────────────────────────────────────────────────────

#[derive(Debug)]
pub enum CryptoError {
    /// No DEK provisioned for this `KeyId`.
    UnknownKey(KeyId),
    /// The DEK (or this version of it) was destroyed — the data is erased
    /// by definition. Callers should surface this as "gone", not "error".
    Shredded(KeyId),
    /// AEAD failure: wrong key, corrupted ciphertext, or tampering.
    Aead,
    /// Malformed `enc:v1:` wire string.
    WireFormat,
    /// The KEK source failed (network, auth, KMS outage).
    Kek(Box<dyn std::error::Error + Send + Sync>),
    /// (De)serialization of a field value failed.
    Codec(serde_json::Error),
}

impl fmt::Display for CryptoError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnknownKey(k) => write!(f, "no data key provisioned for `{k}`"),
            Self::Shredded(k) => write!(f, "data key `{k}` has been shredded — data is erased"),
            Self::Aead => write!(f, "AEAD failure: wrong key or tampered ciphertext"),
            Self::WireFormat => write!(f, "malformed encrypted-field wire format"),
            Self::Kek(e) => write!(f, "KEK source error: {e}"),
            Self::Codec(e) => write!(f, "field codec error: {e}"),
        }
    }
}
impl std::error::Error for CryptoError {}

// ─── Key identity & material ──────────────────────────────────────────────────

/// Identifies one DEK lineage. Conventional scopes:
/// `tenant:<id>` (tenant-wide) and `subject:<id>` (per-person, shreddable).
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct KeyId(pub SmolStr);

impl KeyId {
    pub fn new(id: impl AsRef<str>) -> Self {
        Self(SmolStr::new(id.as_ref()))
    }
    pub fn tenant(id: impl AsRef<str>) -> Self {
        Self(SmolStr::new(format!("tenant:{}", id.as_ref())))
    }
    pub fn subject(id: impl AsRef<str>) -> Self {
        Self(SmolStr::new(format!("subject:{}", id.as_ref())))
    }
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for KeyId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

/// One unwrapped AES-256 DEK version. Material is zeroized on drop
/// (`secrecy`) and never serialized.
pub struct DataKey {
    version: u32,
    material: Secret<[u8; 32]>,
}

impl DataKey {
    pub fn new(version: u32, material: [u8; 32]) -> Self {
        Self {
            version,
            material: Secret::new(material),
        }
    }
    pub fn version(&self) -> u32 {
        self.version
    }
}

// ─── KEK source — the only I/O boundary ──────────────────────────────────────

/// Bridges to the key-management system. Production implementations wrap
/// Vault Transit / AWS KMS / GCP KMS; dev uses an env-derived source.
/// Every method runs on the control plane — never on a request path.
/// A full key ring as loaded from the KMS: every key with all live versions.
pub type LoadedKeyring = Vec<(KeyId, Vec<DataKey>)>;

pub trait KekSource: Send + Sync + 'static {
    /// Unwrap and return every (key, all live versions) pair. Called once at
    /// boot; the result seeds the in-memory ring.
    fn load_keyring(&self) -> BoxFuture<'_, Result<LoadedKeyring, CryptoError>>;

    /// Create + persist (wrapped) the next version of `id` — version 1 if
    /// the key is new. Returns the unwrapped DEK.
    fn provision(&self, id: &KeyId) -> BoxFuture<'_, Result<DataKey, CryptoError>>;

    /// Destroy every wrapped version of `id` permanently. After this returns
    /// the data sealed under `id` is unrecoverable from any sink.
    fn destroy(&self, id: &KeyId) -> BoxFuture<'_, Result<(), CryptoError>>;
}

// ─── Wire format ──────────────────────────────────────────────────────────────

/// Self-describing ciphertext for one field:
/// `enc:v1:<key_id>:<key_version>:<b64(nonce ‖ ciphertext ‖ tag)>`.
/// Records the key version so rotation never forces immediate re-encryption.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct EncryptedField {
    pub key_id: KeyId,
    pub key_version: u32,
    /// 12-byte GCM nonce followed by ciphertext+tag.
    pub blob: Vec<u8>,
}

impl EncryptedField {
    pub fn to_wire(&self) -> String {
        format!(
            "{WIRE_PREFIX}{}:{}:{}",
            self.key_id,
            self.key_version,
            B64.encode(&self.blob)
        )
    }

    /// Parse the wire form. `key_id` may itself contain `:` (e.g.
    /// `tenant:acme`), so version and blob are taken from the *right*.
    pub fn from_wire(s: &str) -> Result<Self, CryptoError> {
        let rest = s.strip_prefix(WIRE_PREFIX).ok_or(CryptoError::WireFormat)?;
        let mut it = rest.rsplitn(3, ':');
        let blob_b64 = it.next().ok_or(CryptoError::WireFormat)?;
        let version = it
            .next()
            .and_then(|v| v.parse::<u32>().ok())
            .ok_or(CryptoError::WireFormat)?;
        let key_id = it
            .next()
            .filter(|k| !k.is_empty())
            .ok_or(CryptoError::WireFormat)?;
        let blob = B64.decode(blob_b64).map_err(|_| CryptoError::WireFormat)?;
        if blob.len() < 12 + 16 {
            return Err(CryptoError::WireFormat);
        }
        Ok(Self {
            key_id: KeyId::new(key_id),
            key_version: version,
            blob,
        })
    }

    /// Cheap check sinks can use to tell sealed values from plaintext.
    pub fn is_wire(s: &str) -> bool {
        s.starts_with(WIRE_PREFIX)
    }
}

// ─── Key ring snapshot ────────────────────────────────────────────────────────

/// Immutable snapshot of every unwrapped DEK. Replaced whole on any control
/// plane change; readers always see a consistent ring.
struct KeyRingSnapshot {
    /// Versions sorted ascending; the last one is active (used to encrypt).
    keys: HashMap<KeyId, Vec<DataKey>>,
    epoch: u64,
}

impl KeyRingSnapshot {
    fn active_key(&self, id: &KeyId) -> Option<&DataKey> {
        self.keys.get(id).and_then(|v| v.last())
    }
    fn key_version(&self, id: &KeyId, version: u32) -> Option<&DataKey> {
        self.keys
            .get(id)
            .and_then(|v| v.iter().find(|k| k.version == version))
    }
}

// ─── The vault ────────────────────────────────────────────────────────────────

/// Process-wide encryption service. Provide once via
/// `ctx.provide(CryptoVault::bootstrap(source).await?)`, resolve anywhere
/// with `Inject<CryptoVault>` / `ctx.inject::<CryptoVault>()`.
pub struct CryptoVault {
    ring: ArcSwap<KeyRingSnapshot>,
    source: Arc<dyn KekSource>,
    /// Serializes control-plane rebuilds (provision / rotate / shred).
    /// Never touched by `encrypt` / `decrypt` — the hot path stays lock-free.
    rebuild: tokio::sync::Mutex<()>,
}

impl CryptoVault {
    /// Load every DEK from the KEK source and build the initial ring.
    pub async fn bootstrap(source: Arc<dyn KekSource>) -> Result<Self, CryptoError> {
        let mut keys: HashMap<KeyId, Vec<DataKey>> = HashMap::new();
        for (id, mut versions) in source.load_keyring().await? {
            versions.sort_by_key(|k| k.version);
            keys.insert(id, versions);
        }
        Ok(Self {
            ring: ArcSwap::from_pointee(KeyRingSnapshot { keys, epoch: 0 }),
            source,
            rebuild: tokio::sync::Mutex::new(()),
        })
    }

    /// Hot path: seal `plaintext` under the active version of `key`.
    /// One atomic load + one hash probe + AES-256-GCM. No locks, no I/O.
    pub fn encrypt(&self, key: &KeyId, plaintext: &[u8]) -> Result<EncryptedField, CryptoError> {
        let ring = self.ring.load();
        let dk = ring
            .active_key(key)
            .ok_or_else(|| CryptoError::UnknownKey(key.clone()))?;
        let cipher = Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(dk.material.expose_secret()));
        let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
        let ct = cipher
            .encrypt(&nonce, plaintext)
            .map_err(|_| CryptoError::Aead)?;
        let mut blob = Vec::with_capacity(12 + ct.len());
        blob.extend_from_slice(&nonce);
        blob.extend_from_slice(&ct);
        Ok(EncryptedField {
            key_id: key.clone(),
            key_version: dk.version,
            blob,
        })
    }

    /// Hot path: open a sealed field with the exact version that sealed it.
    /// Returns `Shredded` when the key (or version) is gone — by design.
    pub fn decrypt(&self, field: &EncryptedField) -> Result<Secret<Vec<u8>>, CryptoError> {
        let ring = self.ring.load();
        let dk = ring
            .key_version(&field.key_id, field.key_version)
            .ok_or_else(|| CryptoError::Shredded(field.key_id.clone()))?;
        let cipher = Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(dk.material.expose_secret()));
        let (nonce, ct) = field.blob.split_at(12);
        let pt = cipher
            .decrypt(Nonce::from_slice(nonce), ct)
            .map_err(|_| CryptoError::Aead)?;
        Ok(Secret::new(pt))
    }

    /// `true` when an active DEK exists for `key`.
    pub fn has_key(&self, key: &KeyId) -> bool {
        self.ring.load().active_key(key).is_some()
    }

    /// Control plane: provision the key if absent (idempotent). Use for
    /// per-subject keys minted on first write.
    pub async fn ensure_key(&self, key: &KeyId) -> Result<(), CryptoError> {
        if self.has_key(key) {
            return Ok(());
        }
        let _g = self.rebuild.lock().await;
        if self.has_key(key) {
            return Ok(()); // raced with another provisioner
        }
        let dk = self.source.provision(key).await?;
        self.swap_ring(|keys| {
            keys.entry(key.clone()).or_default().push(dk);
        });
        Ok(())
    }

    /// Control plane: add a new active version. Old versions keep decrypting
    /// existing ciphertext; new writes seal under the new version.
    /// Returns the new active version number.
    pub async fn rotate(&self, key: &KeyId) -> Result<u32, CryptoError> {
        let _g = self.rebuild.lock().await;
        let dk = self.source.provision(key).await?;
        let v = dk.version;
        self.swap_ring(|keys| {
            let versions = keys.entry(key.clone()).or_default();
            versions.push(dk);
            versions.sort_by_key(|k| k.version);
        });
        Ok(v)
    }

    /// Control plane: **crypto-shredding**. Destroys the wrapped DEK at the
    /// KEK source, then drops it from the ring. Every ciphertext sealed
    /// under `key` — in the DB, outbox, idempotency cache, DLQ, audit chain
    /// — is permanently unreadable the moment this returns.
    pub async fn shred(&self, key: &KeyId) -> Result<(), CryptoError> {
        let _g = self.rebuild.lock().await;
        self.source.destroy(key).await?;
        self.swap_ring(|keys| {
            keys.remove(key);
        });
        tracing::info!(key = %key, "data key shredded — subject data erased");
        Ok(())
    }

    /// Clone-and-swap the ring snapshot. Caller holds the rebuild mutex.
    fn swap_ring(&self, mutate: impl FnOnce(&mut HashMap<KeyId, Vec<DataKey>>)) {
        let cur = self.ring.load();
        // DataKey isn't Clone (secret material) — rebuild by re-wrapping the
        // secrets we already hold in memory.
        let mut keys: HashMap<KeyId, Vec<DataKey>> = cur
            .keys
            .iter()
            .map(|(k, vs)| {
                (
                    k.clone(),
                    vs.iter()
                        .map(|d| DataKey::new(d.version, *d.material.expose_secret()))
                        .collect(),
                )
            })
            .collect();
        mutate(&mut keys);
        self.ring.store(Arc::new(KeyRingSnapshot {
            keys,
            epoch: cur.epoch + 1,
        }));
    }
}

// ─── JSON field walker ────────────────────────────────────────────────────────

/// One segment of a compiled field path. `*` fans out over arrays/objects,
/// mirroring the masking path engine.
#[derive(Clone, Debug)]
enum Seg {
    Key(&'static str),
    Any,
}

fn compile(spec: &'static str) -> Vec<Seg> {
    spec.split('.')
        .map(|s| if s == "*" { Seg::Any } else { Seg::Key(s) })
        .collect()
}

/// Seal every leaf matched by `path`. The leaf's full JSON encoding is
/// encrypted (types round-trip exactly), and the leaf is replaced by the
/// wire string.
fn seal_at(
    vault: &CryptoVault,
    key: &KeyId,
    v: &mut Value,
    path: &[Seg],
) -> Result<(), CryptoError> {
    match path.split_first() {
        None => {
            let plain = serde_json::to_vec(v).map_err(CryptoError::Codec)?;
            *v = Value::String(vault.encrypt(key, &plain)?.to_wire());
            Ok(())
        }
        Some((Seg::Key(k), rest)) => match v.get_mut(*k) {
            Some(child) => seal_at(vault, key, child, rest),
            None => Ok(()), // absent field: nothing to seal
        },
        Some((Seg::Any, rest)) => {
            match v {
                Value::Array(items) => {
                    for item in items {
                        seal_at(vault, key, item, rest)?;
                    }
                }
                Value::Object(map) => {
                    for child in map.values_mut() {
                        seal_at(vault, key, child, rest)?;
                    }
                }
                _ => {}
            }
            Ok(())
        }
    }
}

/// Inverse of [`seal_at`]: decrypt matched leaves that carry the wire prefix.
fn unseal_at(vault: &CryptoVault, v: &mut Value, path: &[Seg]) -> Result<(), CryptoError> {
    match path.split_first() {
        None => {
            let Value::String(s) = &*v else { return Ok(()) };
            if !EncryptedField::is_wire(s) {
                return Ok(()); // already plaintext (e.g. pre-rollout rows)
            }
            let field = EncryptedField::from_wire(s)?;
            let plain = vault.decrypt(&field)?;
            *v = serde_json::from_slice(plain.expose_secret()).map_err(CryptoError::Codec)?;
            Ok(())
        }
        Some((Seg::Key(k), rest)) => match v.get_mut(*k) {
            Some(child) => unseal_at(vault, child, rest),
            None => Ok(()),
        },
        Some((Seg::Any, rest)) => {
            match v {
                Value::Array(items) => {
                    for item in items {
                        unseal_at(vault, item, rest)?;
                    }
                }
                Value::Object(map) => {
                    for child in map.values_mut() {
                        unseal_at(vault, child, rest)?;
                    }
                }
                _ => {}
            }
            Ok(())
        }
    }
}

// ─── Declarative record contract (#[EncryptFields]) ──────────────────────────

/// Implemented by `#[EncryptFields(...)]` on a DTO. The default methods are
/// the whole write/read contract:
///
/// - **seal**: serialize → encrypt the declared fields → `Value` safe for
///   any sink (DB row, outbox payload, idempotency cache, DLQ).
/// - **unseal**: decrypt the declared fields → deserialize back to `Self`.
///
/// `KEY_ID` is the default scope; use the `*_with_key` variants for
/// per-subject keys resolved at runtime.
pub trait EncryptRecord: serde::Serialize + serde::de::DeserializeOwned {
    /// Dotted paths (`"card.number"`, `"items.*.ssn"`) to seal.
    const ENCRYPT_FIELDS: &'static [&'static str];
    /// Default key scope, e.g. `"tenant:acme"`.
    const KEY_ID: &'static str;

    fn seal(&self, vault: &CryptoVault) -> Result<Value, CryptoError> {
        self.seal_with_key(vault, &KeyId::new(Self::KEY_ID))
    }

    fn seal_with_key(&self, vault: &CryptoVault, key: &KeyId) -> Result<Value, CryptoError> {
        let mut v = serde_json::to_value(self).map_err(CryptoError::Codec)?;
        for spec in Self::ENCRYPT_FIELDS {
            seal_at(vault, key, &mut v, &compile(spec))?;
        }
        Ok(v)
    }

    fn unseal(mut sealed: Value, vault: &CryptoVault) -> Result<Self, CryptoError> {
        for spec in Self::ENCRYPT_FIELDS {
            unseal_at(vault, &mut sealed, &compile(spec))?;
        }
        serde_json::from_value(sealed).map_err(CryptoError::Codec)
    }
}

// ─── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use sha2::{Digest, Sha256};

    /// Deterministic in-memory source: DEK = SHA-256(master ‖ key ‖ version).
    struct TestKek {
        shredded: std::sync::Mutex<std::collections::HashSet<KeyId>>,
        versions: std::sync::Mutex<HashMap<KeyId, u32>>,
    }

    impl TestKek {
        fn new() -> Self {
            Self {
                shredded: Default::default(),
                versions: Default::default(),
            }
        }
        fn derive(id: &KeyId, version: u32) -> [u8; 32] {
            let mut h = Sha256::new();
            h.update(b"test-master");
            h.update(id.as_str().as_bytes());
            h.update(version.to_be_bytes());
            h.finalize().into()
        }
    }

    impl KekSource for TestKek {
        fn load_keyring(&self) -> BoxFuture<'_, Result<Vec<(KeyId, Vec<DataKey>)>, CryptoError>> {
            Box::pin(async { Ok(Vec::new()) })
        }
        fn provision(&self, id: &KeyId) -> BoxFuture<'_, Result<DataKey, CryptoError>> {
            let id = id.clone();
            Box::pin(async move {
                let mut versions = self.versions.lock().unwrap();
                let v = versions.entry(id.clone()).or_insert(0);
                *v += 1;
                Ok(DataKey::new(*v, Self::derive(&id, *v)))
            })
        }
        fn destroy(&self, id: &KeyId) -> BoxFuture<'_, Result<(), CryptoError>> {
            let id = id.clone();
            Box::pin(async move {
                self.shredded.lock().unwrap().insert(id);
                Ok(())
            })
        }
    }

    async fn vault() -> CryptoVault {
        CryptoVault::bootstrap(Arc::new(TestKek::new()))
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn roundtrip_and_wire_format() {
        let v = vault().await;
        let key = KeyId::tenant("acme");
        v.ensure_key(&key).await.unwrap();

        let sealed = v.encrypt(&key, b"4242-4242").unwrap();
        let wire = sealed.to_wire();
        assert!(EncryptedField::is_wire(&wire));

        let parsed = EncryptedField::from_wire(&wire).unwrap();
        assert_eq!(parsed, sealed);
        assert_eq!(parsed.key_id, key); // key id with ':' survives the wire

        let plain = v.decrypt(&parsed).unwrap();
        assert_eq!(plain.expose_secret().as_slice(), b"4242-4242");
    }

    #[tokio::test]
    async fn rotation_keeps_old_ciphertext_readable() {
        let v = vault().await;
        let key = KeyId::tenant("acme");
        v.ensure_key(&key).await.unwrap();

        let old = v.encrypt(&key, b"before-rotation").unwrap();
        let new_version = v.rotate(&key).await.unwrap();
        assert_eq!(new_version, 2);

        // old ciphertext still opens; new writes use v2
        assert_eq!(
            v.decrypt(&old).unwrap().expose_secret().as_slice(),
            b"before-rotation"
        );
        assert_eq!(v.encrypt(&key, b"x").unwrap().key_version, 2);
    }

    #[tokio::test]
    async fn shred_makes_data_unrecoverable() {
        let v = vault().await;
        let key = KeyId::subject("user-42");
        v.ensure_key(&key).await.unwrap();
        let sealed = v.encrypt(&key, b"phi").unwrap();

        v.shred(&key).await.unwrap();
        assert!(matches!(v.decrypt(&sealed), Err(CryptoError::Shredded(_))));
        assert!(matches!(
            v.encrypt(&key, b"more"),
            Err(CryptoError::UnknownKey(_))
        ));
    }

    #[tokio::test]
    async fn tampered_ciphertext_fails_aead() {
        let v = vault().await;
        let key = KeyId::tenant("acme");
        v.ensure_key(&key).await.unwrap();
        let mut sealed = v.encrypt(&key, b"secret").unwrap();
        *sealed.blob.last_mut().unwrap() ^= 0xFF;
        assert!(matches!(v.decrypt(&sealed), Err(CryptoError::Aead)));
    }

    #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
    struct Patient {
        name: String,
        ssn: String,
        visits: Vec<Visit>,
    }
    #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
    struct Visit {
        diagnosis: String,
        year: u32,
    }

    impl EncryptRecord for Patient {
        const ENCRYPT_FIELDS: &'static [&'static str] = &["ssn", "visits.*.diagnosis"];
        const KEY_ID: &'static str = "tenant:clinic";
    }

    #[tokio::test]
    async fn record_seal_unseal_with_wildcards() {
        let v = vault().await;
        v.ensure_key(&KeyId::new("tenant:clinic")).await.unwrap();

        let p = Patient {
            name: "Jane".into(),
            ssn: "123-45-6789".into(),
            visits: vec![
                Visit {
                    diagnosis: "A".into(),
                    year: 2024,
                },
                Visit {
                    diagnosis: "B".into(),
                    year: 2025,
                },
            ],
        };

        let sealed = p.seal(&v).unwrap();
        // declared fields are wire strings; everything else is plaintext
        assert!(EncryptedField::is_wire(sealed["ssn"].as_str().unwrap()));
        assert!(EncryptedField::is_wire(
            sealed["visits"][0]["diagnosis"].as_str().unwrap()
        ));
        assert_eq!(sealed["name"], "Jane");
        assert_eq!(sealed["visits"][1]["year"], 2025);

        let back = Patient::unseal(sealed, &v).unwrap();
        assert_eq!(back, p);
    }

    #[tokio::test]
    async fn unseal_tolerates_pre_rollout_plaintext() {
        let v = vault().await;
        v.ensure_key(&KeyId::new("tenant:clinic")).await.unwrap();
        // a row written before encryption was enabled
        let legacy = serde_json::json!({
            "name": "Old", "ssn": "raw-ssn", "visits": []
        });
        let p = Patient::unseal(legacy, &v).unwrap();
        assert_eq!(p.ssn, "raw-ssn");
    }
}