hyphae-engine 0.1.0

// SPDX-License-Identifier: Apache-2.0

use std::{
    fs::{File, OpenOptions},
    io::{Read, Write},
    path::Path,
    time::Instant,
};

use hyphae_query::{Record, execute};
use hyphae_storage::load_snapshot;

use super::{
    ProofAnchor, ProofError, ProvenOperation, ProvenResult, ResultProof, VerificationLimits,
    VerificationReport, decode_proof, encode_proof,
};
use crate::decode_document;

/// Writes a canonical result proof to a new file and synchronizes it.
///
/// Existing paths are never replaced.
///
/// # Errors
///
/// Returns a proof encoding, path, create, write, or synchronization error.
pub fn write_result_proof(path: impl AsRef<Path>, proof: &ResultProof) -> Result<(), ProofError> {
    let encoded = encode_proof(proof)?;
    let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
    file.write_all(&encoded)?;
    file.sync_all()?;
    Ok(())
}

/// Reads and verifies one canonical result-proof file under a byte limit.
///
/// # Errors
///
/// Returns an I/O, resource-limit, framing, canonicality, checksum, or digest
/// error.
pub fn read_result_proof(
    path: impl AsRef<Path>,
    maximum_bytes: u64,
) -> Result<ResultProof, ProofError> {
    let mut file = File::open(path)?;
    let metadata_length = file.metadata()?.len();
    if metadata_length > maximum_bytes {
        return Err(ProofError::ProofLimitExceeded {
            actual: metadata_length,
            maximum: maximum_bytes,
        });
    }
    let capacity = usize::try_from(metadata_length).map_err(|_| ProofError::LengthOverflow)?;
    let mut encoded = Vec::with_capacity(capacity);
    file.read_to_end(&mut encoded)?;
    let actual = u64::try_from(encoded.len()).map_err(|_| ProofError::LengthOverflow)?;
    if actual > maximum_bytes {
        return Err(ProofError::ProofLimitExceeded {
            actual,
            maximum: maximum_bytes,
        });
    }
    if actual != metadata_length {
        return Err(ProofError::Invalid {
            reason: "proof changed while being read",
        });
    }
    decode_proof(&encoded)
}

/// Verifies a result proof completely offline against a trusted anchor and
/// canonical snapshot witness.
///
/// # Errors
///
/// Returns an error for any proof or snapshot corruption, wrong anchor,
/// resource exhaustion, document failure, timeout, or replay mismatch. No
/// partial result is accepted.
pub fn verify_result_proof(
    proof_path: impl AsRef<Path>,
    snapshot_path: impl AsRef<Path>,
    expected_anchor_digest: [u8; 32],
    limits: &VerificationLimits,
) -> Result<VerificationReport, ProofError> {
    let started = Instant::now();
    let proof = read_result_proof(proof_path, limits.proof_bytes)?;
    check_timeout(started, limits)?;

    let anchor_digest = proof.anchor_digest();
    if anchor_digest != expected_anchor_digest {
        return Err(ProofError::AnchorMismatch);
    }

    let snapshot = load_snapshot(snapshot_path, &limits.snapshot)?;
    check_timeout(started, limits)?;
    if ProofAnchor::from_snapshot(&snapshot.info) != *proof.anchor() {
        return Err(ProofError::SnapshotAnchorMismatch);
    }

    let mut records = Vec::with_capacity(snapshot.entries.len());
    for entry in snapshot.entries {
        check_timeout(started, limits)?;
        records.push(Record {
            key: entry.key,
            value: decode_document(&entry.value)?,
        });
    }

    let verified_result = match (proof.operation(), proof.result()) {
        (ProvenOperation::Get { key }, ProvenResult::Get(expected)) => {
            let actual = records
                .binary_search_by(|record| record.key.as_slice().cmp(key))
                .ok()
                .map(|index| records[index].clone());
            if &actual != expected {
                return Err(ProofError::ReexecutionMismatch);
            }
            ProvenResult::Get(actual)
        }
        (ProvenOperation::Query(query), ProvenResult::Query(expected)) => {
            let elapsed = started.elapsed();
            let remaining = limits
                .timeout
                .checked_sub(elapsed)
                .ok_or(ProofError::TimedOut)?;
            if remaining.is_zero() {
                return Err(ProofError::TimedOut);
            }
            let query_limits = hyphae_query::ExecutionLimits {
                timeout: remaining.min(limits.query.timeout),
                ..limits.query.clone()
            };
            let actual = execute(&[records.as_slice()], query, &query_limits)?;
            if &actual != expected {
                return Err(ProofError::ReexecutionMismatch);
            }
            ProvenResult::Query(actual)
        }
        _ => return Err(ProofError::OperationResultMismatch),
    };
    check_timeout(started, limits)?;

    Ok(VerificationReport {
        anchor: proof.anchor().clone(),
        anchor_digest,
        proof_digest: proof.proof_digest(),
        result: verified_result,
    })
}

fn check_timeout(started: Instant, limits: &VerificationLimits) -> Result<(), ProofError> {
    if started.elapsed() >= limits.timeout {
        Err(ProofError::TimedOut)
    } else {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::BTreeMap, error::Error, fs, path::PathBuf};

    use hyphae_query::{ExecutionLimits, Filter, Query, Record, Value};
    use uuid::Uuid;

    use super::{VerificationLimits, verify_result_proof, write_result_proof};
    use crate::{HyphaeEngine, ProofError, ProvenResult};

    struct TestDirectory {
        path: PathBuf,
    }

    impl TestDirectory {
        fn create() -> Result<Self, Box<dyn Error>> {
            let path = std::env::temp_dir()
                .join(format!("hyphae-proof-rehashed-tamper-{}", Uuid::now_v7()));
            fs::create_dir_all(&path)?;
            Ok(Self { path })
        }
    }

    impl Drop for TestDirectory {
        fn drop(&mut self) {
            let _ignored = fs::remove_dir_all(&self.path);
        }
    }

    #[test]
    fn self_consistently_rehashed_result_edits_are_rejected() -> Result<(), Box<dyn Error>> {
        let temporary = TestDirectory::create()?;
        let mut opened = HyphaeEngine::open(temporary.path.join("data"))?;
        opened.engine.put_records(
            Uuid::now_v7(),
            &[record(b"a", 1), record(b"b", 2), record(b"c", 3)],
        )?;
        let artifact = opened.engine.query_with_proof(
            &Query {
                filter: Filter::MatchAll,
                sort: Vec::new(),
                cursor: None,
                limit: 3,
                aggregation: None,
            },
            &ExecutionLimits::default(),
        )?;

        for (name, mutation) in [
            ("delete", 1_u8),
            ("insert", 2_u8),
            ("reorder", 3_u8),
            ("edit", 4_u8),
        ] {
            let mut forged = artifact.proof.clone();
            let ProvenResult::Query(result) = &mut forged.result else {
                return Err(ProofError::OperationResultMismatch.into());
            };
            match mutation {
                1 => {
                    result.rows.remove(0);
                }
                2 => result.rows.push(result.rows[0].clone()),
                3 => result.rows.swap(0, 1),
                4 => result.rows[0].value = Value::Integer(999),
                _ => return Err(ProofError::OperationResultMismatch.into()),
            }
            let proof_path = temporary.path.join(format!("{name}.hyproof"));
            write_result_proof(&proof_path, &forged)?;
            assert!(matches!(
                verify_result_proof(
                    &proof_path,
                    &artifact.snapshot.path,
                    artifact.proof.anchor_digest(),
                    &VerificationLimits::default(),
                ),
                Err(ProofError::ReexecutionMismatch)
            ));
        }
        Ok(())
    }

    fn record(key: &[u8], score: i64) -> Record {
        Record::new(
            key,
            Value::Object(BTreeMap::from([(
                "score".to_owned(),
                Value::Integer(score),
            )])),
        )
    }
}