vti-common 0.45.0

//! Task-execution consent — **DTTE** (Delegated Trust-Task Execution) — the
//! data layer behind the PDP's `requireConsent` disposition.
//!
//! "DTTE" is the name this subsystem goes by in prose and in design notes; the
//! code spells it `task_consent` / `requireConsent` throughout. Operator
//! documentation is `docs/02-vta/task-consent.md`.
//!
//! Not to be confused with *messaging consent* (`consent/*/1.0`, the
//! `consent_ks` keyspace), which asks whether two parties may correspond.
//!
//! When a policy returns `requireConsent`, a privileged task can't proceed until
//! one or more named approvers have signed off on **this exact task**. The
//! binding is a deterministic digest of the task URI and its payload (RFC 8785
//! JCS + SHA-256), so the approval a re-submitted task consumes provably
//! concerns the same request — an approver can't be tricked into signing one
//! payload while a different one executes, nor into approving a benign task URI
//! whose payload happens to canonicalize like a destructive one's.
//!
//! Two records in the `task_consent` keyspace:
//! - [`PendingTaskConsent`] (`pending:<digest>`) — an in-flight request
//!   accumulating approver signatures.
//! - [`TaskConsentGrant`] (`grant:<digest>:<requester>`) — a completed
//!   authorization the re-submitting requester consumes single-use.
//!
//! This mirrors the step-up "reject → approve → re-submit" loop, but the
//! authorization is bound to the payload digest rather than the session.
//!
//! **Node-neutral.** Every node type that runs the ceremony — the VTA's PDP
//! gate and the VTC's unrestricted-admin gate (VTI-APV-014) — keeps its
//! pending requests and grants through this module, so the digest, the
//! single-use consume and the expiry rules cannot drift between them. What a
//! node gates on, who its approvers are and how it pushes the request stay
//! with the node. `vta_policy::consent` re-exports this module.

pub mod effects;

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::error::AppError;
use crate::store::{Batch, KeyspaceHandle};

const PENDING_PREFIX: &[u8] = b"pending:";
const GRANT_PREFIX: &[u8] = b"grant:";
const WIRE_INDEX_PREFIX: &[u8] = b"wire:";

/// Domain-separation tag: keeps these digests from colliding with any other
/// SHA-256 over a canonical payload elsewhere in the system.
///
/// The `vta/` prefix predates the module's move out of `vta-policy` and is
/// kept deliberately: the digest keys stored pending requests and grants, so
/// changing the tag would orphan every ceremony in flight across an upgrade.
/// It separates this digest from other digests, not one node from another —
/// pendings and grants never leave the node that minted them.
const DIGEST_DOMAIN: &[u8] = b"vta/task-consent/v1\0";

/// Deterministic digest of a task: SHA-256 over the type URI and the RFC 8785
/// (JCS) canonical payload, encoded as a multibase multihash
/// ([`encode_digest_multibase`]). Stable across serializers, so the requester's
/// re-submit and the approver's signed decision agree on what was authorized.
///
/// The type URI is **part of the digest**, and is length-prefixed so the
/// URI/payload boundary can't be shifted. Without it, two tasks whose payloads
/// canonicalize identically — `{"did":…,"contextId":…}` is a plausible payload
/// for `dids/update`, `dids/rotate-keys` *and* a deactivate — would share a
/// digest, and an approval for the benign one would authorize the destructive
/// one. The approver only ever sees an opaque digest, so nothing downstream
/// could catch the substitution.
/// This digest is **executor-internal**. It never leaves the process: it is the
/// key under which a pending request and its grant are stored, and the gate must
/// be able to recompute it on a re-submit *before* it knows the challenge — which
/// is precisely why it cannot be the salted one. See [`wire_digest`].
pub fn payload_digest(type_uri: &str, payload: &serde_json::Value) -> Result<String, AppError> {
    digest_with(type_uri, payload, None)
}

/// The digest the approver signs, the requester echoes, and the two screens
/// match: [`payload_digest`] salted with the per-request `challenge`.
///
/// Salted because an unsalted digest over a low-entropy payload is a confirmation
/// oracle. "Deactivate `did:webvh:abc…`" has essentially one canonical form, so
/// anyone who observes the digest in transit — a compromised, subpoenaed, or
/// retrospectively-decrypted mediator — can guess the operation and hash to check.
/// Both screens still derive the same value because both hold the challenge.
///
/// The salt is why there are two digests rather than one. Keying storage by the
/// salted value would be circular: the gate would need the challenge to compute
/// the key under which the challenge is stored.
pub fn wire_digest(
    type_uri: &str,
    payload: &serde_json::Value,
    challenge: &str,
) -> Result<String, AppError> {
    digest_with(type_uri, payload, Some(challenge))
}

fn digest_with(
    type_uri: &str,
    payload: &serde_json::Value,
    challenge: Option<&str>,
) -> Result<String, AppError> {
    domain_digest(DIGEST_DOMAIN, type_uri, payload, challenge)
}

/// The construction under [`payload_digest`] and [`wire_digest`], for another
/// domain: SHA-256 over `domain`, the length-prefixed type URI, the
/// length-prefixed RFC 8785 canonical payload and, when given, the salt — as a
/// base58btc multibase multihash.
///
/// For a digest that binds one operation the same way but must never be
/// mistaken for a task-consent digest: the VTC's operation-bound step-up mark
/// (`vtc/step-up/v1\0`) is one. `domain` should end in a NUL so no tag is a
/// prefix of another.
pub fn domain_digest(
    domain: &[u8],
    type_uri: &str,
    payload: &serde_json::Value,
    salt: Option<&str>,
) -> Result<String, AppError> {
    let canonical = serde_json_canonicalizer::to_string(payload)
        .map_err(|e| AppError::Internal(format!("payload JCS canonicalization failed: {e}")))?;
    let mut h = Sha256::new();
    h.update(domain);
    h.update((type_uri.len() as u64).to_be_bytes());
    h.update(type_uri.as_bytes());
    h.update((canonical.len() as u64).to_be_bytes());
    h.update(canonical.as_bytes());
    if let Some(c) = salt {
        h.update(c.as_bytes());
    }
    Ok(encode_digest_multibase(&h.finalize()))
}

/// Multihash prefix for SHA-256 with a 32-byte digest: code `0x12`, length
/// `0x20`. Carried in-band so the value is self-describing and the wire format
/// survives an algorithm change without a schema revision.
const MULTIHASH_SHA2_256_32: [u8; 2] = [0x12, 0x20];

/// Encode a SHA-256 digest as a multibase-encoded multihash — the
/// `digestMultibase` form W3C VCDM 2.0 defines and `did:webvh` uses.
///
/// This replaced bare hex when the Trust Tasks registry moved `payloadDigest`
/// to the shared `DigestMultibase` type, which explicitly rules out "a bare hex
/// string or a `sha-256:`-style prefix" as non-conforming. **This is a wire
/// change, not a rendering change**: the digest is what an approver signs and
/// what the requester's re-submit is matched against, so every party that
/// computes one independently has to move together. See
/// `docs/05-design-notes/approvals-convergence.md`.
///
/// base58btc (`z`) for consistency with `did:key` and `did:webvh`, matching
/// `vta_sdk::did_key::ed25519_multibase_pubkey`.
///
/// **That choice is now load-bearing rather than stylistic.** The type's
/// accepted set narrowed in `trust-tasks-rs` 0.6.0 to the two headers W3C
/// Controlled Identifiers 1.0 §2.4 normatively requires — `z` (base58btc) and
/// `u` (base64url-no-pad) — with each alphabet enforced rather than assumed.
/// The earlier pattern (`^[zumbfF]…`) also admitted base32/base16/base64pad,
/// so a peer that had picked `b` or `f` would parse then and fail now. We were
/// already on `z`; keep it, and do not "generalise" this to another header.
fn encode_digest_multibase(digest: &[u8]) -> String {
    let mut buf = Vec::with_capacity(MULTIHASH_SHA2_256_32.len() + digest.len());
    buf.extend_from_slice(&MULTIHASH_SHA2_256_32);
    buf.extend_from_slice(digest);
    multibase::encode(multibase::Base::Base58Btc, &buf)
}

/// An in-flight consent request accumulating approver signatures.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PendingTaskConsent {
    /// Internal [`payload_digest`] of the task awaiting consent.
    pub digest: String,
    /// The salted [`wire_digest`] the approver signs. Stored so a decision can be
    /// matched without re-deriving it from a payload we deliberately do not keep.
    pub wire_digest: String,
    /// Type URI of the task (for the approver's display + audit).
    pub type_uri: String,
    /// The DID that submitted the task.
    pub requester_did: String,
    /// Named approver set the policy required (resolved to members at check time).
    pub approver_set: String,
    /// Distinct approvals needed before a grant is issued.
    pub min_approvals: u32,
    /// When true, the requester's own DID cannot count toward the threshold.
    pub exclude_requester: bool,
    /// Nonce the approver echoes + signs, binding the decision to this request.
    pub challenge: String,
    /// Freshly-minted correlator for this ceremony, used as the `threadId` on
    /// the documents it produces.
    ///
    /// Framework 0.5.0 (*Identifier correlation and linkability*) requires
    /// `id`, `threadId` and `ceremony.enactment` to be "freshly minted and
    /// unguessable" and **MUST NOT** be derived from subject data. The granted
    /// notice used to thread on [`wire_digest`], which is derived from the task
    /// payload — salted, so not recoverable, but still a function of the
    /// content and the same string the documents carry as `payloadDigest`.
    ///
    /// A mediator sees `threadId` as routing metadata. With the digest there it
    /// could link the refusal, the approval pushes and the notice into one
    /// ceremony *and* tie them to the payload digest it forwards, which is the
    /// linkage the rule exists to remove.
    ///
    /// `#[serde(default)]` so a pending written before this field existed still
    /// loads; such a record correlates on an empty string, which is no worse
    /// than the notice being unthreaded.
    #[serde(default)]
    pub correlator: String,
    /// Distinct approver DIDs who have approved so far.
    pub approvals: Vec<String>,
    /// The prior state the effects shown to the approver were computed against.
    ///
    /// The payload itself is **not** stored — the requester re-submits it, and the
    /// digest proves it is the same one. What must be kept is the state the
    /// effects were computed against, because that is what execution has to
    /// re-assert: a human in the loop makes the window minutes wide, so the world
    /// can move underneath an approval.
    #[serde(default)]
    pub state_pin: Option<effects::StatePin>,
    /// Executor-internal preconditions to re-assert at execution.
    #[serde(default)]
    pub guards: crate::guards::Guards,
    /// The context whose admin authority the task acts under (webvh update: the
    /// DID's context), when the planner could determine it. An approver must
    /// administer this context for their approval to confer execution authority.
    #[serde(default)]
    pub subject_context: Option<String>,
    /// Whether the requester's own token authorized `subject_context`. `false`
    /// marks a cross-context proposal whose execution requires a delegation from
    /// a context-admin approver. Defaults `true` so pre-feature pendings (and
    /// tasks with no context subject) are never treated as delegated.
    #[serde(default = "default_true")]
    pub requester_authorized: bool,
    pub created_at: u64,
    pub expires_at: u64,
}

fn default_true() -> bool {
    true
}

/// A completed authorization a re-submitted task consumes (single-use).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskConsentGrant {
    pub digest: String,
    pub requester_did: String,
    pub type_uri: String,
    /// The approver DIDs whose signatures produced this grant.
    pub approvers: Vec<String>,
    /// Carried from the pending request: what the approvers were shown, and what
    /// execution must still find true.
    #[serde(default)]
    pub state_pin: Option<effects::StatePin>,
    #[serde(default)]
    pub guards: crate::guards::Guards,
    /// Contexts this grant confers admin authority in for the single execution
    /// that consumes it. Empty for an ordinary same-context consent (the
    /// requester already held the context); populated only when the approvals
    /// came from admins of a context the requester lacked — the per-task
    /// delegation. Consumed by widening the executing `AuthClaims` via
    /// `AuthClaims::with_delegated_contexts`, never persisted.
    #[serde(default)]
    pub delegated_contexts: Vec<String>,
    pub granted_at: u64,
    pub expires_at: u64,
}

fn pending_key(digest: &str) -> Vec<u8> {
    [PENDING_PREFIX, digest.as_bytes()].concat()
}

/// Index from the salted wire digest back to the internal one.
///
/// The approver's decision carries only the wire digest — it is the only digest
/// that ever left the process — but the pending request is keyed by the internal
/// one, because the gate has to find it on a re-submit before it knows the
/// challenge. This index closes that loop.
fn wire_index_key(wire_digest: &str) -> Vec<u8> {
    [WIRE_INDEX_PREFIX, wire_digest.as_bytes()].concat()
}

fn grant_key(requester_did: &str, digest: &str) -> Vec<u8> {
    // `:` can't appear in a hex digest; the requester DID may contain `:`, so put
    // it last after the fixed-shape prefix+digest to keep the key unambiguous.
    [
        GRANT_PREFIX,
        digest.as_bytes(),
        b":",
        requester_did.as_bytes(),
    ]
    .concat()
}

fn decode<T: for<'de> Deserialize<'de>>(bytes: &[u8]) -> Result<T, AppError> {
    serde_json::from_slice(bytes)
        .map_err(|e| AppError::Internal(format!("task-consent decode: {e}")))
}

// ── Pending ────────────────────────────────────────────────────────────────

/// How many times a read-modify-write of one pending request re-reads it after
/// another writer changed it between the read and the conditional write.
///
/// An attempt is lost only to another decision on the same request that was
/// applied, so a writer loses at most once per concurrent approver; the bound
/// only stops a pathological stream from holding a caller forever.
const WRITE_ATTEMPTS: usize = 64;

/// Store a pending request and its wire index, as one store batch — an index
/// without its pending (or the reverse) would leave a decision unresolvable.
pub async fn store_pending(ks: &KeyspaceHandle, p: &PendingTaskConsent) -> Result<(), AppError> {
    let mut batch = Batch::new();
    batch.insert(pending_key(&p.digest), p)?;
    // Index the wire digest, since that is the only one an approver's decision
    // can carry back to us.
    batch.insert_raw(wire_index_key(&p.wire_digest), p.digest.as_bytes().to_vec());
    ks.apply_batch(batch).await?;
    Ok(())
}

/// Resolve the salted wire digest an approver signed back to the internal digest
/// the pending request is keyed by.
pub async fn pending_by_wire_digest(
    ks: &KeyspaceHandle,
    wire_digest: &str,
    now: u64,
) -> Result<Option<PendingTaskConsent>, AppError> {
    let Some(bytes) = ks.get_raw(wire_index_key(wire_digest)).await? else {
        return Ok(None);
    };
    let internal = String::from_utf8(bytes)
        .map_err(|e| AppError::Internal(format!("wire index value not utf-8: {e}")))?;
    get_pending(ks, &internal, now).await
}

/// Fetch a live pending consent. An expired one is treated as absent and swept
/// — its TTL is what bounds how long an approver's signature can authorize a
/// request, so it has to be enforced on the read path, not only by the sweeper.
pub async fn get_pending(
    ks: &KeyspaceHandle,
    digest: &str,
    now: u64,
) -> Result<Option<PendingTaskConsent>, AppError> {
    Ok(read_live_pending(ks, digest, now).await?.map(|(_, p)| p))
}

/// The live pending at `digest` as stored (plaintext, for a `require_unchanged`
/// precondition) and as parsed. Sweeps an expired one — only while it is still
/// the expired row read, so a fresh request re-stored at the same digest in the
/// meantime is not swept with it.
async fn read_live_pending(
    ks: &KeyspaceHandle,
    digest: &str,
    now: u64,
) -> Result<Option<(Vec<u8>, PendingTaskConsent)>, AppError> {
    let key = pending_key(digest);
    let Some(b) = ks.get_raw(key.clone()).await? else {
        return Ok(None);
    };
    let p: PendingTaskConsent = decode(&b)?;
    if p.expires_at <= now {
        let mut batch = Batch::new();
        batch
            .require_unchanged(key.clone(), b)
            .remove(key)
            .remove(wire_index_key(&p.wire_digest));
        ks.apply_batch(batch).await?;
        return Ok(None);
    }
    Ok(Some((b, p)))
}

/// Delete a pending request and its wire index, as one store batch. Takes the
/// record rather than a digest so both keys are always removed together — an
/// orphaned index would resolve to a pending that no longer exists.
pub async fn delete_pending(ks: &KeyspaceHandle, p: &PendingTaskConsent) -> Result<(), AppError> {
    let mut batch = Batch::new();
    batch
        .remove(pending_key(&p.digest))
        .remove(wire_index_key(&p.wire_digest));
    ks.apply_batch(batch).await?;
    Ok(())
}

/// Record an approval (idempotent per approver) and return the updated pending.
/// `Ok(None)` if there is no live pending consent for the digest. The caller
/// decides whether `approvals.len() >= min_approvals` and, if so, issues a grant
/// with [`grant_from_pending`].
///
/// The write applies only while the pending is still the row read, so it can
/// neither lose a concurrent approver's signature nor resurrect a request a
/// concurrent denial (or grant) deleted — a resurrected pending would let a
/// later approval reach the threshold of a request that was refused.
pub async fn add_approval(
    ks: &KeyspaceHandle,
    digest: &str,
    approver_did: &str,
    now: u64,
) -> Result<Option<PendingTaskConsent>, AppError> {
    for _ in 0..WRITE_ATTEMPTS {
        let Some((raw, mut p)) = read_live_pending(ks, digest, now).await? else {
            return Ok(None);
        };
        if p.approvals.iter().any(|a| a == approver_did) {
            return Ok(Some(p));
        }
        p.approvals.push(approver_did.to_string());
        let key = pending_key(digest);
        let mut batch = Batch::new();
        batch.require_unchanged(key.clone(), raw).insert(key, &p)?;
        if ks.apply_batch(batch).await?.is_applied() {
            return Ok(Some(p));
        }
    }
    Err(AppError::Conflict(format!(
        "task-consent pending {digest} kept changing while an approval was recorded"
    )))
}

/// Mint `grant` from the pending request it completes and retire that request,
/// as one store batch: the grant is written, and the pending and its wire index
/// removed, only while the pending is still live and still carries every
/// approval the grant names.
///
/// Returns `false` (nothing written) when the pending is gone — a concurrent
/// denial deleted it, another approval already minted the grant, or it lapsed.
/// Writing the grant and deleting the pending separately let a denial landing
/// between an approval and the grant be overridden by the grant, and two
/// approvals crossing the threshold together each mint one.
pub async fn grant_from_pending(
    ks: &KeyspaceHandle,
    grant: &TaskConsentGrant,
    now: u64,
) -> Result<bool, AppError> {
    for _ in 0..WRITE_ATTEMPTS {
        let Some((raw, p)) = read_live_pending(ks, &grant.digest, now).await? else {
            return Ok(false);
        };
        if p.requester_did != grant.requester_did
            || p.type_uri != grant.type_uri
            || !grant.approvers.iter().all(|a| p.approvals.contains(a))
            || (p.approvals.len() as u32) < p.min_approvals
        {
            return Ok(false);
        }
        let key = pending_key(&p.digest);
        let mut batch = Batch::new();
        batch
            .require_unchanged(key.clone(), raw)
            .insert(grant_key(&grant.requester_did, &grant.digest), grant)?
            .remove(key)
            .remove(wire_index_key(&p.wire_digest));
        if ks.apply_batch(batch).await?.is_applied() {
            return Ok(true);
        }
    }
    Err(AppError::Conflict(format!(
        "task-consent pending {} kept changing while its grant was minted",
        grant.digest
    )))
}

/// Prune expired pendings and lapsed grants. Both paths already expire lazily on
/// read; this bounds the space an unanswered request can hold indefinitely.
///
/// Each removal applies only while the row is still the expired one read, so a
/// request or grant re-stored at the same key meanwhile survives.
pub async fn sweep_expired(ks: &KeyspaceHandle, now: u64) -> Result<usize, AppError> {
    let mut pruned = 0usize;

    for (key, value) in ks.prefix_iter_raw("pending:").await? {
        match serde_json::from_slice::<PendingTaskConsent>(&value) {
            Ok(p) if p.expires_at <= now => {
                let mut batch = Batch::new();
                batch
                    .require_unchanged(key.clone(), value)
                    .remove(key)
                    .remove(wire_index_key(&p.wire_digest));
                if ks.apply_batch(batch).await?.is_applied() {
                    pruned += 1;
                }
            }
            Ok(_) => {}
            Err(e) => tracing::debug!(error = %e, "task-consent sweeper: unreadable pending row"),
        }
    }

    for (key, value) in ks.prefix_iter_raw("grant:").await? {
        match serde_json::from_slice::<TaskConsentGrant>(&value) {
            Ok(g) if g.expires_at <= now => {
                let mut batch = Batch::new();
                batch.require_unchanged(key.clone(), value).remove(key);
                if ks.apply_batch(batch).await?.is_applied() {
                    pruned += 1;
                }
            }
            Ok(_) => {}
            Err(e) => tracing::debug!(error = %e, "task-consent sweeper: unreadable grant row"),
        }
    }

    Ok(pruned)
}

/// Remove a pending request and its wire index only if the pending is still
/// stored, as one store batch. Returns whether it was there to remove — `false`
/// means a concurrent decision already resolved it (an approval minted its
/// grant, or another denial), so the caller must not report it as denied.
pub async fn remove_pending_if_present(
    ks: &KeyspaceHandle,
    p: &PendingTaskConsent,
) -> Result<bool, AppError> {
    let mut batch = Batch::new();
    batch
        .require_present(pending_key(&p.digest))
        .remove(pending_key(&p.digest))
        .remove(wire_index_key(&p.wire_digest));
    Ok(ks.apply_batch(batch).await?.is_applied())
}

// ── Grant ──────────────────────────────────────────────────────────────────

pub async fn store_grant(ks: &KeyspaceHandle, g: &TaskConsentGrant) -> Result<(), AppError> {
    ks.insert(grant_key(&g.requester_did, &g.digest), g).await
}

/// Consume a valid grant for `(requester, digest)` — single-use: on a hit the
/// grant is removed before returning. Returns `None` if absent or expired (an
/// expired grant is also removed). This is the gate's allow-path check.
///
/// The read and the removal are one atomic take, so of two re-submits racing
/// on one grant exactly one receives it.
///
/// `type_uri` is re-asserted against the grant even though [`payload_digest`]
/// already folds it in: the digest binding is the load-bearing defence, and this
/// is the assertion that fails loudly if that ever regresses.
pub async fn consume_grant(
    ks: &KeyspaceHandle,
    requester_did: &str,
    type_uri: &str,
    digest: &str,
    now: u64,
) -> Result<Option<TaskConsentGrant>, AppError> {
    // Remove either way: a hit is single-use, an expired grant is swept.
    let Some(bytes) = ks.take_raw(grant_key(requester_did, digest)).await? else {
        return Ok(None);
    };
    let grant: TaskConsentGrant = decode(&bytes)?;
    if grant.expires_at <= now {
        return Ok(None);
    }
    if grant.type_uri != type_uri {
        return Err(AppError::Internal(format!(
            "task-consent grant type mismatch: granted for '{}', presented for '{type_uri}'",
            grant.type_uri
        )));
    }
    Ok(Some(grant))
}

/// Read the live grant for `(requester, digest)` **without spending it**.
///
/// For a gate that must find two things before it may spend either — the VTC
/// needs both another admin's consent and the requester's own bound passkey
/// gesture, and must not burn one while the other is missing. The caller spends
/// with [`consume_grant`], which is still the single-use step: a grant this read
/// returns may be gone by the time the caller consumes it, and the caller must
/// treat that as absent. An expired grant is removed and reads as absent.
pub async fn get_grant(
    ks: &KeyspaceHandle,
    requester_did: &str,
    digest: &str,
    now: u64,
) -> Result<Option<TaskConsentGrant>, AppError> {
    let key = grant_key(requester_did, digest);
    let Some(bytes) = ks.get_raw(key.clone()).await? else {
        return Ok(None);
    };
    let grant: TaskConsentGrant = decode(&bytes)?;
    if grant.expires_at <= now {
        // Only the expired row read: a fresh grant stored meanwhile stays.
        let mut batch = Batch::new();
        batch.require_unchanged(key.clone(), bytes).remove(key);
        ks.apply_batch(batch).await?;
        return Ok(None);
    }
    Ok(Some(grant))
}

/// Remove the grant for `(requester, digest)`, if any — for a gate that has read
/// one with [`get_grant`] and found it no longer authorizes anything.
pub async fn discard_grant(
    ks: &KeyspaceHandle,
    requester_did: &str,
    digest: &str,
) -> Result<(), AppError> {
    ks.remove(grant_key(requester_did, digest)).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::StoreConfig;
    use crate::store::Store;
    use serde_json::json;

    async fn temp_ks() -> (KeyspaceHandle, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(&StoreConfig {
            data_dir: dir.path().to_path_buf(),
        })
        .unwrap();
        (store.keyspace("task_consent").unwrap(), dir)
    }

    const T_UPDATE: &str = "https://trusttasks.org/spec/webvh/dids/update/1.0";
    const T_ROTATE: &str = "https://trusttasks.org/spec/webvh/dids/rotate-keys/1.0";

    #[test]
    fn digest_is_deterministic_and_key_order_independent() {
        let a = payload_digest(T_UPDATE, &json!({ "b": 2, "a": 1 })).unwrap();
        let b = payload_digest(T_UPDATE, &json!({ "a": 1, "b": 2 })).unwrap();
        assert_eq!(a, b, "JCS canonicalization must ignore key order");
        assert_ne!(
            a,
            payload_digest(T_UPDATE, &json!({ "a": 1, "b": 3 })).unwrap()
        );
        // `digestMultibase`, not hex: base58btc (`z`) over the sha2-256
        // multihash. Asserted by shape rather than length, because base58 is
        // not byte-aligned and a leading-zero digest encodes shorter.
        assert!(a.starts_with('z'), "must be multibase base58btc, got {a}");
        let (base, bytes) = multibase::decode(&a).expect("valid multibase");
        assert_eq!(base, multibase::Base::Base58Btc);
        assert_eq!(
            &bytes[..2],
            &[0x12, 0x20],
            "must carry the sha2-256 multihash prefix in-band"
        );
        assert_eq!(bytes.len(), 34, "2-byte multihash prefix + 32-byte digest");
    }

    /// The bypass this binding exists to close: an identical payload under two
    /// different task URIs must not share a digest, or an approval for the
    /// benign task authorizes the destructive one.
    #[test]
    fn digest_binds_the_type_uri() {
        let payload = json!({ "did": "did:webvh:example.com:abc", "contextId": "default" });
        assert_ne!(
            payload_digest(T_UPDATE, &payload).unwrap(),
            payload_digest(T_ROTATE, &payload).unwrap(),
            "same payload under a different task URI must not collide"
        );
    }

    /// The wire digest is the one that leaves the process, and it is salted with
    /// the challenge so it cannot be used as a confirmation oracle: a payload
    /// like "deactivate did:webvh:abc…" has essentially one canonical form, so an
    /// unsalted digest is guessable by anyone who observes it in transit.
    #[test]
    fn wire_digest_is_salted_and_distinct_from_the_internal_one() {
        let p = json!({ "did": "did:webvh:example.com:acme" });
        let internal = payload_digest(T_UPDATE, &p).unwrap();
        let a = wire_digest(T_UPDATE, &p, "challenge-aaaa").unwrap();
        let b = wire_digest(T_UPDATE, &p, "challenge-bbbb").unwrap();

        assert_ne!(
            internal, a,
            "the wire digest must not equal the storage key"
        );
        assert_ne!(
            a, b,
            "a different challenge must yield a different wire digest"
        );
        // …and it is still deterministic for a given challenge, or the approver's
        // echo could never be matched.
        assert_eq!(a, wire_digest(T_UPDATE, &p, "challenge-aaaa").unwrap());
    }

    /// The wire digest binds the type URI too — the same reason the internal one
    /// does, and the approver only ever sees this one.
    #[test]
    fn wire_digest_binds_the_type_uri() {
        let p = json!({ "did": "did:webvh:example.com:acme" });
        assert_ne!(
            wire_digest(T_UPDATE, &p, "c").unwrap(),
            wire_digest(T_ROTATE, &p, "c").unwrap()
        );
    }

    /// The decision an approver signs carries only the wire digest, so the
    /// pending request has to be reachable from it.
    #[tokio::test]
    async fn a_decision_resolves_its_pending_by_wire_digest() {
        let (ks, _d) = temp_ks().await;
        let p = pending("deadbeef", 1);
        store_pending(&ks, &p).await.unwrap();

        let found = pending_by_wire_digest(&ks, &p.wire_digest, 200)
            .await
            .unwrap()
            .expect("resolved via the wire index");
        assert_eq!(found.digest, "deadbeef");

        assert!(
            pending_by_wire_digest(&ks, "not-a-digest", 200)
                .await
                .unwrap()
                .is_none()
        );

        // Deleting the pending must take its index with it — an orphaned index
        // would resolve to a request that no longer exists.
        delete_pending(&ks, &p).await.unwrap();
        assert!(
            pending_by_wire_digest(&ks, &p.wire_digest, 200)
                .await
                .unwrap()
                .is_none()
        );
    }

    /// Length-prefixing the URI stops a boundary shift between the URI and the
    /// canonical payload from producing the same preimage.
    #[test]
    fn digest_uri_payload_boundary_is_unambiguous() {
        assert_ne!(
            payload_digest("ab", &json!("c")).unwrap(),
            payload_digest("a", &json!("bc")).unwrap(),
        );
    }

    /// Fixed vectors, computed independently of this code. The digest keys
    /// stored pendings and grants and is what approvers sign, so a change to
    /// the construction — even one every round-trip test above would still
    /// pass — must fail here first.
    #[test]
    fn digest_matches_its_pinned_vectors() {
        const URI: &str = "https://trusttasks.org/spec/acl/grant/0.1";
        let p = json!({ "role": "admin", "did": "did:key:z6MkA" });
        assert_eq!(
            payload_digest(URI, &p).unwrap(),
            "zQmNsLmSgtT4jrcgWEC8nHhjambJmen2phxzMKVrCoDWL4Q"
        );
        assert_eq!(
            wire_digest(URI, &p, "chal").unwrap(),
            "zQmYdqiqiHsozX3N5NRnM6CmYRwWmPo4P5Pu1MVopiCfwa2"
        );
        assert_eq!(
            domain_digest(b"vtc/step-up/v1\0", URI, &p, None).unwrap(),
            "zQmcC7Puan5QbvK5HZ5HWSzrfr7m9pLk8ajR6JRfvB2BG7k"
        );
        assert_eq!(
            domain_digest(b"vtc/step-up/v1\0", URI, &p, Some("chal")).unwrap(),
            "zQmc1qnh2HJTpPzHCsPV81Dred9WPgqRKBooU7WYZdGqJEi"
        );
    }

    fn pending(digest: &str, min: u32) -> PendingTaskConsent {
        PendingTaskConsent {
            digest: digest.into(),
            wire_digest: format!("wire-{digest}"),
            correlator: "urn:uuid:test-correlator".into(),
            state_pin: None,
            guards: Default::default(),
            type_uri: "https://…/dids/update/1.0".into(),
            requester_did: "did:key:zReq".into(),
            approver_set: "operators".into(),
            min_approvals: min,
            exclude_requester: true,
            challenge: "nonce123".into(),
            approvals: vec![],
            subject_context: None,
            requester_authorized: true,
            created_at: 100,
            expires_at: 1000,
        }
    }

    #[tokio::test]
    async fn approvals_accumulate_idempotently() {
        let (ks, _d) = temp_ks().await;
        store_pending(&ks, &pending("deadbeef", 2)).await.unwrap();

        let p = add_approval(&ks, "deadbeef", "did:key:zA", 200)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(p.approvals.len(), 1);
        // Same approver again → no double count.
        let p = add_approval(&ks, "deadbeef", "did:key:zA", 200)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(p.approvals.len(), 1);
        // Second distinct approver reaches the threshold.
        let p = add_approval(&ks, "deadbeef", "did:key:zB", 200)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(p.approvals.len(), 2);
        assert!(p.approvals.len() as u32 >= p.min_approvals);

        // No pending for an unknown digest.
        assert!(
            add_approval(&ks, "nope", "did:key:zA", 200)
                .await
                .unwrap()
                .is_none()
        );
    }

    /// `expires_at` was previously written but never read, so a pending consent
    /// could be approved indefinitely.
    #[tokio::test]
    async fn expired_pending_reads_as_absent_and_cannot_be_approved() {
        let (ks, _d) = temp_ks().await;
        store_pending(&ks, &pending("deadbeef", 1)).await.unwrap();

        // expires_at = 1000
        assert!(get_pending(&ks, "deadbeef", 999).await.unwrap().is_some());
        assert!(get_pending(&ks, "deadbeef", 1000).await.unwrap().is_none());
        assert!(
            add_approval(&ks, "deadbeef", "did:key:zA", 1001)
                .await
                .unwrap()
                .is_none(),
            "an expired pending must not accept approvals"
        );
        // …and the lapsed row is swept on the way out.
        assert!(get_pending(&ks, "deadbeef", 500).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn sweeper_prunes_lapsed_pendings_and_grants() {
        let (ks, _d) = temp_ks().await;
        store_pending(&ks, &pending("d1", 1)).await.unwrap(); // expires_at 1000
        store_grant(&ks, &grant("d2", 500)).await.unwrap();

        assert_eq!(
            sweep_expired(&ks, 400).await.unwrap(),
            0,
            "nothing lapsed yet"
        );
        assert_eq!(sweep_expired(&ks, 2000).await.unwrap(), 2, "both lapsed");
        assert_eq!(sweep_expired(&ks, 2000).await.unwrap(), 0, "idempotent");
    }

    fn grant(digest: &str, expires_at: u64) -> TaskConsentGrant {
        TaskConsentGrant {
            digest: digest.into(),
            state_pin: None,
            guards: Default::default(),
            requester_did: "did:key:zReq".into(),
            type_uri: T_UPDATE.into(),
            approvers: vec!["did:key:zA".into()],
            delegated_contexts: vec![],
            granted_at: 100,
            expires_at,
        }
    }

    #[tokio::test]
    async fn grant_is_single_use_and_expiry_checked() {
        let (ks, _d) = temp_ks().await;
        let g = grant("d1", 500);
        store_grant(&ks, &g).await.unwrap();

        // First consume within validity → hit.
        assert!(
            consume_grant(&ks, "did:key:zReq", T_UPDATE, "d1", 200)
                .await
                .unwrap()
                .is_some()
        );
        // Second consume → gone (single-use).
        assert!(
            consume_grant(&ks, "did:key:zReq", T_UPDATE, "d1", 200)
                .await
                .unwrap()
                .is_none()
        );

        // Expired grant → None, and swept.
        store_grant(&ks, &g).await.unwrap();
        assert!(
            consume_grant(&ks, "did:key:zReq", T_UPDATE, "d1", 999)
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            consume_grant(&ks, "did:key:zReq", T_UPDATE, "d1", 200)
                .await
                .unwrap()
                .is_none()
        );

        // Wrong requester never matches.
        store_grant(&ks, &g).await.unwrap();
        assert!(
            consume_grant(&ks, "did:key:zOther", T_UPDATE, "d1", 200)
                .await
                .unwrap()
                .is_none()
        );
    }

    /// A read that does not spend: the grant is still there to consume, and an
    /// expired one reads as absent and is gone.
    #[tokio::test]
    async fn get_grant_reads_without_spending() {
        let (ks, _d) = temp_ks().await;
        store_grant(&ks, &grant("d1", 500)).await.unwrap();

        assert!(
            get_grant(&ks, "did:key:zReq", "d1", 200)
                .await
                .unwrap()
                .is_some()
        );
        assert!(
            get_grant(&ks, "did:key:zReq", "d1", 200)
                .await
                .unwrap()
                .is_some()
        );
        assert!(
            get_grant(&ks, "did:key:zOther", "d1", 200)
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            consume_grant(&ks, "did:key:zReq", T_UPDATE, "d1", 200)
                .await
                .unwrap()
                .is_some(),
            "a read must leave the grant for the consume"
        );

        store_grant(&ks, &grant("d1", 500)).await.unwrap();
        assert!(
            get_grant(&ks, "did:key:zReq", "d1", 600)
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            get_grant(&ks, "did:key:zReq", "d1", 200)
                .await
                .unwrap()
                .is_none(),
            "an expired grant is removed on read"
        );

        store_grant(&ks, &grant("d1", 500)).await.unwrap();
        discard_grant(&ks, "did:key:zReq", "d1").await.unwrap();
        assert!(
            get_grant(&ks, "did:key:zReq", "d1", 200)
                .await
                .unwrap()
                .is_none()
        );
    }

    /// Belt-and-braces on the type binding: even if a digest for one task URI
    /// were somehow presented for another, consumption must refuse it rather
    /// than silently authorize.
    #[tokio::test]
    async fn grant_refuses_a_different_task_uri() {
        let (ks, _d) = temp_ks().await;
        store_grant(&ks, &grant("d1", 500)).await.unwrap();

        let err = consume_grant(&ks, "did:key:zReq", T_ROTATE, "d1", 200)
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("type mismatch"),
            "expected a type-mismatch refusal, got: {err}"
        );
    }

    // ── Concurrency: every multi-step write above is one conditional batch ──

    const ROUNDS: usize = 40;
    const RACERS: usize = 16;

    /// Single use under contention: of many re-submits racing on one grant,
    /// exactly one receives it. A get-then-remove handed it to several.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn concurrent_consumes_spend_a_grant_exactly_once() {
        let (ks, _d) = temp_ks().await;
        for _ in 0..ROUNDS {
            store_grant(&ks, &grant("d1", 500)).await.unwrap();
            let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(RACERS));
            let mut tasks = Vec::new();
            for _ in 0..RACERS {
                let ks = ks.clone();
                let barrier = barrier.clone();
                tasks.push(tokio::spawn(async move {
                    barrier.wait().await;
                    consume_grant(&ks, "did:key:zReq", T_UPDATE, "d1", 200)
                        .await
                        .unwrap()
                        .is_some()
                }));
            }
            let mut hits = 0;
            for t in tasks {
                hits += t.await.unwrap() as usize;
            }
            assert_eq!(hits, 1, "a grant must be spent exactly once");
        }
    }

    /// Distinct approvers deciding together must all be counted. An unlocked
    /// read-then-store let each overwrite the others' signatures.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn concurrent_approvals_are_never_lost() {
        let (ks, _d) = temp_ks().await;
        for round in 0..ROUNDS {
            let digest = format!("d{round}");
            store_pending(&ks, &pending(&digest, RACERS as u32))
                .await
                .unwrap();
            let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(RACERS));
            let mut tasks = Vec::new();
            for i in 0..RACERS {
                let (ks, barrier, digest) = (ks.clone(), barrier.clone(), digest.clone());
                tasks.push(tokio::spawn(async move {
                    barrier.wait().await;
                    add_approval(&ks, &digest, &format!("did:key:z{i}"), 200)
                        .await
                        .unwrap()
                }));
            }
            for t in tasks {
                t.await.unwrap();
            }
            let p = get_pending(&ks, &digest, 200).await.unwrap().unwrap();
            assert_eq!(p.approvals.len(), RACERS, "an approval was lost");
        }
    }

    /// A denial is final: an approval racing it must not write the pending back,
    /// or a later approval could reach the threshold of a refused request.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn an_approval_never_resurrects_a_denied_pending() {
        let (ks, _d) = temp_ks().await;
        for round in 0..ROUNDS * 4 {
            let p = pending(&format!("d{round}"), 2);
            store_pending(&ks, &p).await.unwrap();
            let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(RACERS + 1));
            let mut tasks = Vec::new();
            for i in 0..RACERS {
                let (ks, barrier, digest) = (ks.clone(), barrier.clone(), p.digest.clone());
                tasks.push(tokio::spawn(async move {
                    barrier.wait().await;
                    add_approval(&ks, &digest, &format!("did:key:z{i}"), 200)
                        .await
                        .unwrap();
                }));
            }
            let deny = {
                let (ks, barrier, p) = (ks.clone(), barrier.clone(), p.clone());
                tokio::spawn(async move {
                    barrier.wait().await;
                    delete_pending(&ks, &p).await.unwrap();
                })
            };
            for t in tasks {
                t.await.unwrap();
            }
            deny.await.unwrap();
            assert!(
                get_pending(&ks, &p.digest, 200).await.unwrap().is_none(),
                "a denied pending came back"
            );
        }
    }

    /// Of a denial and the approvals that cross the threshold together, exactly
    /// one decision resolves the request: either one grant is minted and the
    /// denial finds nothing, or the denial lands and no grant exists.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn a_pending_resolves_to_exactly_one_grant_or_one_denial() {
        let (ks, _d) = temp_ks().await;
        for round in 0..ROUNDS {
            let digest = format!("d{round}");
            let mut p = pending(&digest, 1);
            p.approvals = vec!["did:key:zA".into()];
            store_pending(&ks, &p).await.unwrap();
            let mut g = grant(&digest, 5000);
            g.type_uri = p.type_uri.clone();

            let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(RACERS + 1));
            let mut grants = Vec::new();
            for _ in 0..RACERS {
                let (ks, barrier, g) = (ks.clone(), barrier.clone(), g.clone());
                grants.push(tokio::spawn(async move {
                    barrier.wait().await;
                    grant_from_pending(&ks, &g, 200).await.unwrap()
                }));
            }
            let deny = {
                let (ks, barrier, p) = (ks.clone(), barrier.clone(), p.clone());
                tokio::spawn(async move {
                    barrier.wait().await;
                    remove_pending_if_present(&ks, &p).await.unwrap()
                })
            };
            let mut minted = 0;
            for t in grants {
                minted += t.await.unwrap() as usize;
            }
            let denied = deny.await.unwrap() as usize;
            assert_eq!(minted + denied, 1, "minted {minted}, denied {denied}");
            let granted = get_grant(&ks, "did:key:zReq", &digest, 200)
                .await
                .unwrap()
                .is_some();
            assert_eq!(granted, minted == 1, "a grant exists iff one was minted");
            assert!(get_pending(&ks, &digest, 200).await.unwrap().is_none());
            assert!(
                pending_by_wire_digest(&ks, &p.wire_digest, 200)
                    .await
                    .unwrap()
                    .is_none(),
                "the wire index must go with its pending"
            );
        }
    }

    /// A pending and its wire index are written and retired together: one
    /// journal record each time, so a crash never leaves one without the other.
    #[tokio::test]
    async fn pending_and_its_index_commit_as_one_journal_record() {
        let (ks, _d) = temp_ks().await;
        let seq = || ks.journal_seqno().unwrap();
        let mut p = pending("d1", 1);
        p.approvals = vec!["did:key:zA".into()];
        let before = seq();
        store_pending(&ks, &p).await.unwrap();
        assert_eq!(seq() - before, 1, "store_pending");
        let mut g = grant("d1", 5000);
        g.type_uri = p.type_uri.clone();
        let before = seq();
        assert!(grant_from_pending(&ks, &g, 200).await.unwrap());
        assert_eq!(seq() - before, 1, "grant_from_pending");
    }

    /// A grant is minted only from the approvals it names.
    #[tokio::test]
    async fn grant_from_pending_refuses_unrecorded_approvers() {
        let (ks, _d) = temp_ks().await;
        let mut p = pending("d1", 1);
        p.approvals = vec!["did:key:zA".into()];
        store_pending(&ks, &p).await.unwrap();
        let mut g = grant("d1", 5000);
        g.type_uri = p.type_uri.clone();
        g.approvers = vec!["did:key:zB".into()];
        assert!(!grant_from_pending(&ks, &g, 200).await.unwrap());
        assert!(get_pending(&ks, "d1", 200).await.unwrap().is_some());
        g.approvers = vec!["did:key:zA".into()];
        assert!(grant_from_pending(&ks, &g, 200).await.unwrap());
        assert!(get_pending(&ks, "d1", 200).await.unwrap().is_none());
    }
}