newton-core 0.5.4

//! Content-addressed object resolution over a [`UserStorageBackend`].
//!
//! Objects live in the configured backend under `objects/<cid>` and every read
//! is integrity-checked before the bytes are returned.

use std::{fmt::Display, sync::Arc};

use cid::{multihash::Multihash, Cid};
use eyre::{Context, Result};
use tracing::instrument;

use super::backend::UserStorageBackend;
use crate::cid::{cid_v1_raw_cid, parse_persisted_cid, sha2_256_multihash, verify_raw_cid, DAG_PB_CODEC, RAW_CODEC};

/// Backend prefix under which content-addressed objects are stored.
const OBJECTS_PREFIX: &str = "objects";
/// Trusted SHA2-256 multihash sidecars for migrated DAG-PB file bytes.
const OBJECT_CHECKSUMS_PREFIX: &str = "object-checksums";

/// Resolve content-addressed objects (WASM, Rego, schemas) by CID.
///
/// Raw payloads are verified directly against their CID. DAG-PB payloads are
/// verified against the checksum sidecar created by the explicit migration.
#[derive(Clone)]
pub struct PersistedImmutableDataResolver {
    backend: Arc<dyn UserStorageBackend>,
}

impl std::fmt::Debug for PersistedImmutableDataResolver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PersistedImmutableDataResolver").finish_non_exhaustive()
    }
}

impl PersistedImmutableDataResolver {
    /// Build a resolver over `backend`.
    pub fn new(backend: Arc<dyn UserStorageBackend>) -> Self {
        Self { backend }
    }

    /// Backend key for a content-addressed object: `objects/<cid>`.
    pub fn object_key(cid: impl Display) -> String {
        format!("{OBJECTS_PREFIX}/{cid}")
    }

    /// Backend key for a DAG-PB object's SHA2-256 multihash sidecar.
    pub fn checksum_key(cid: impl Display) -> String {
        format!("{OBJECT_CHECKSUMS_PREFIX}/{cid}.sha256")
    }

    /// Resolve `cid` to its caller-facing content.
    pub async fn get(&self, cid: &str) -> Result<Vec<u8>> {
        let parsed = parse_persisted_cid(cid).map_err(|error| eyre::eyre!(error))?;
        self.get_cid(&parsed).await
    }

    /// Resolve an already-parsed `cid` to its caller-facing content.
    pub async fn get_cid(&self, cid: &Cid) -> Result<Vec<u8>> {
        self.get_optional(cid)
            .await?
            .ok_or_else(|| eyre::eyre!("object {cid} not found in persisted immutable data backend"))
    }

    /// Probe the configured backend, distinguishing absence from failure.
    #[instrument(
        name = "storage.object.fetch",
        target = "newton::task_evaluation",
        level = "debug",
        skip(self),
        fields(storage.backend = self.backend.backend_name(), storage.object.cid = %cid)
    )]
    pub async fn get_optional(&self, cid: &Cid) -> Result<Option<Vec<u8>>> {
        let key = Self::object_key(cid);
        let Some(data) = self
            .backend
            .get_optional(&key)
            .await
            .wrap_err_with(|| format!("resolve object {cid}"))?
        else {
            return Ok(None);
        };
        match cid.codec() {
            RAW_CODEC => {
                verify_raw_cid(cid, &data).map_err(|e| eyre::eyre!("integrity check failed for {cid}: {e}"))?;
            }
            DAG_PB_CODEC => {
                let checksum_key = Self::checksum_key(cid);
                let expected = self
                    .backend
                    .get_optional(&checksum_key)
                    .await
                    .wrap_err_with(|| format!("resolve DAG-PB object checksum {cid}"))?
                    .ok_or_else(|| eyre::eyre!("DAG-PB object {cid} is missing its checksum sidecar"))?;
                let expected = Multihash::<64>::from_bytes(&expected)
                    .map_err(|error| eyre::eyre!("invalid checksum multihash for DAG-PB object {cid}: {error}"))?;
                if expected != sha2_256_multihash(&data) {
                    eyre::bail!("integrity check failed for DAG-PB object {cid}: SHA2-256 checksum mismatch");
                }
            }
            codec => eyre::bail!("unsupported CID codec for {cid}: 0x{codec:x}"),
        }

        Ok(Some(data))
    }

    /// Resolve `cid` to UTF-8 text (schemas, Rego source).
    pub async fn get_text(&self, cid: &str) -> Result<String> {
        let data = self.get(cid).await?;
        String::from_utf8(data).wrap_err_with(|| format!("object {cid} is not valid UTF-8"))
    }

    /// Store `data` content-addressed, returning its CIDv1.
    ///
    /// Idempotent: re-storing identical bytes yields the same CID and overwrites
    /// the same key with identical content.
    pub async fn put(&self, data: &[u8]) -> Result<String> {
        let cid = cid_v1_raw_cid(data);
        self.backend.put(&Self::object_key(cid), data).await?;
        Ok(cid.to_string())
    }
}

// ObjectFetcher impl — gated on `rpc` feature to match the trait definition.
#[cfg(feature = "rpc")]
#[async_trait::async_trait]
impl crate::common::task::rpc::ObjectFetcher for PersistedImmutableDataResolver {
    async fn get_object(&self, cid: &str) -> eyre::Result<Vec<u8>> {
        self.get(cid).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::persisted_immutable_data::backend::LocalBackend;

    #[tokio::test]
    async fn put_then_get_roundtrips() {
        let tmp = tempfile::tempdir().unwrap();
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(tmp.path())));

        let cid = resolver.put(b"policy rego source").await.unwrap();
        let got = resolver.get(&cid).await.unwrap();
        assert_eq!(got, b"policy rego source");
    }

    #[tokio::test]
    async fn missing_object_fails_backend_only() {
        let tmp = tempfile::tempdir().unwrap();
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(tmp.path())));
        let cid = crate::cid::cid_v1_raw(b"missing");

        let error = resolver.get(&cid).await.unwrap_err();
        assert!(error.to_string().contains("not found"), "{error}");
    }

    #[tokio::test]
    async fn get_rejects_corrupted_object() {
        let tmp = tempfile::tempdir().unwrap();
        let backend = Arc::new(LocalBackend::new(tmp.path()));
        let resolver = PersistedImmutableDataResolver::new(backend.clone());

        let cid = resolver.put(b"original").await.unwrap();
        let parsed = parse_persisted_cid(&cid).unwrap();
        backend
            .put(&PersistedImmutableDataResolver::object_key(parsed), b"tampered")
            .await
            .unwrap();

        let err = resolver.get(&cid).await.unwrap_err();
        assert!(err.to_string().contains("integrity check failed"), "{err}");
    }

    #[tokio::test]
    async fn get_text_reads_utf8() {
        let tmp = tempfile::tempdir().unwrap();
        let resolver = PersistedImmutableDataResolver::new(Arc::new(LocalBackend::new(tmp.path())));
        let cid = resolver.put("схема".as_bytes()).await.unwrap();
        assert_eq!(resolver.get_text(&cid).await.unwrap(), "схема");
    }

    #[tokio::test]
    async fn dag_pb_object_requires_matching_checksum_sidecar() {
        use sha2::{Digest, Sha256};

        let tmp = tempfile::tempdir().unwrap();
        let backend = Arc::new(LocalBackend::new(tmp.path()));
        let resolver = PersistedImmutableDataResolver::new(backend.clone());
        let payload = b"migrated DAG-PB policy";
        let root_block_digest = Sha256::digest(b"dag-pb root block");
        let multihash = Multihash::<64>::wrap(0x12, root_block_digest.as_slice()).unwrap();
        let cid = Cid::new_v1(crate::cid::DAG_PB_CODEC, multihash);

        backend
            .put(&PersistedImmutableDataResolver::object_key(cid), payload)
            .await
            .unwrap();
        assert!(resolver
            .get(&cid.to_string())
            .await
            .unwrap_err()
            .to_string()
            .contains("checksum sidecar"));

        let checksum = sha2_256_multihash(payload).to_bytes();
        backend
            .put(&PersistedImmutableDataResolver::checksum_key(cid), &checksum)
            .await
            .unwrap();
        assert_eq!(resolver.get(&cid.to_string()).await.unwrap(), payload);

        let wrong_checksum = sha2_256_multihash(b"different payload").to_bytes();
        backend
            .put(&PersistedImmutableDataResolver::checksum_key(cid), &wrong_checksum)
            .await
            .unwrap();
        assert!(resolver
            .get(&cid.to_string())
            .await
            .unwrap_err()
            .to_string()
            .contains("checksum mismatch"));
    }
}