p2panda-stream 0.7.0

// SPDX-License-Identifier: MIT OR Apache-2.0

//! Methods to handle p2panda operations.
use std::borrow::Borrow;

use p2panda_core::prune::validate_prunable_backlink;
use p2panda_core::{
    Extensions, Hash, LogId, Operation, OperationError, SeqNum, VerifyingKey, validate_operation,
};
use p2panda_store::Transaction;
use p2panda_store::logs::LogStore;
use p2panda_store::operations::OperationStore;
use p2panda_store::topics::TopicStore;
use thiserror::Error;

/// Checks an incoming operation for log integrity and persists it into the store when valid.
///
/// Returns true if the operation was inserted to the store, false if the operation is valid but
/// already existed.
pub async fn ingest_operation<S, T, L, E, TP>(
    store: &S,
    operation: &T,
    log_id: &L,
    topic: &TP,
    prune_flag: bool,
) -> Result<bool, IngestError>
where
    S: Transaction
        + OperationStore<Operation<E>, Hash>
        + LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>
        + TopicStore<TP, VerifyingKey, L>,
    T: Borrow<Operation<E>>,
    L: LogId,
    E: Extensions,
{
    let operation: &Operation<E> = operation.borrow();

    // Validate operation format.
    validate_operation(operation)?;

    let permit = store
        .begin()
        .await
        .map_err(|err| IngestError::StoreError(err.to_string()))?;

    // Ignore insertion if operation already exists.
    let already_exists = store
        .has_operation_tx(&operation.hash)
        .await
        .map_err(|err| IngestError::StoreError(err.to_string()))?;

    if already_exists {
        store
            .rollback(permit)
            .await
            .map_err(|err| IngestError::StoreError(err.to_string()))?;

        return Ok(false);
    }

    // Validate log integrity.
    let past_header = store
        .get_latest_entry_tx(&operation.header.verifying_key, log_id)
        .await
        .map_err(|err| IngestError::StoreError(err.to_string()))?
        .map(|operation| operation.header);

    // If no pruning flag is set, we expect the log to have integrity with the previously given
    // operation.
    validate_prunable_backlink(past_header.as_ref(), &operation.header, prune_flag)?;

    // Insert operation into store and associate its log with the given topic.
    let id = operation.hash;
    let verifying_key = operation.header.verifying_key;

    store
        .insert_operation(&id, operation, log_id)
        .await
        .map_err(|err| IngestError::StoreError(err.to_string()))?;

    <S as TopicStore<TP, VerifyingKey, L>>::associate(store, topic, &verifying_key, log_id)
        .await
        .map_err(|err| IngestError::StoreError(err.to_string()))?;

    store
        .commit(permit)
        .await
        .map_err(|err| IngestError::StoreError(err.to_string()))?;

    Ok(true)
}

/// Errors which can occur due to invalid operations or critical storage failures.
#[derive(Clone, Debug, Error)]
pub enum IngestError {
    /// Operation can not be authenticated, has broken log- or payload integrity or doesn't follow
    /// the p2panda specification.
    #[error(transparent)]
    InvalidOperation(#[from] OperationError),

    /// Critical storage failure occurred. This is usually a reason to panic.
    #[error("critical storage failure: {0}")]
    StoreError(String),
}

#[cfg(test)]
mod tests {
    use p2panda_core::test_utils::TestLog;
    use p2panda_core::{Hash, Header, Operation, SeqNum, SigningKey, VerifyingKey};
    use p2panda_store::SqliteStore;
    use p2panda_store::logs::LogStore;
    use p2panda_store::topics::TopicStore;

    use super::ingest_operation;

    #[tokio::test]
    async fn valid_log() {
        let store = SqliteStore::temporary().await;
        let log = TestLog::new();

        for i in 0..128 {
            let operation = log.operation(format!("{i}").as_bytes(), ());
            let result = ingest_operation(&store, &operation, &1, &1, false).await;
            assert!(result.is_ok());
        }
    }

    #[tokio::test]
    async fn deduplicate_operations() {
        let store = SqliteStore::temporary().await;
        let log = TestLog::new();
        let operation = log.operation(b"same same", ());

        let result = ingest_operation(&store, &operation, &1, &1, false)
            .await
            .unwrap();
        assert!(result);

        // Inserting duplicates is ok and are silently ignored.
        let result = ingest_operation(&store, &operation, &1, &1, false)
            .await
            .unwrap();
        assert!(!result);
    }

    #[tokio::test]
    async fn topic_association() {
        let store = SqliteStore::temporary().await;

        let log_0 = TestLog::new();
        let log_1 = TestLog::new();
        let log_2 = TestLog::new();

        let dogs = [2; 32];
        let cats = [3; 32];

        ingest_operation(&store, &log_0.operation(b"Do", ()), &0, &dogs, false)
            .await
            .unwrap();

        ingest_operation(&store, &log_0.operation(b"Re", ()), &0, &dogs, false)
            .await
            .unwrap();

        ingest_operation(&store, &log_1.operation(b"Mi", ()), &1, &dogs, false)
            .await
            .unwrap();

        ingest_operation(&store, &log_2.operation(b"Fa", ()), &2, &cats, false)
            .await
            .unwrap();

        ingest_operation(&store, &log_2.operation(b"So", ()), &2, &cats, false)
            .await
            .unwrap();

        ingest_operation(&store, &log_2.operation(b"La", ()), &2, &cats, false)
            .await
            .unwrap();

        // Topic "dogs" contains two logs: 0 with two operations and 1 with one operation.
        let authors =
            <SqliteStore as TopicStore<[u8; 32], VerifyingKey, usize>>::resolve(&store, &dogs)
                .await
                .unwrap();
        assert_eq!(*authors.get(&log_0.author()).unwrap(), [0]);
        assert_eq!(*authors.get(&log_1.author()).unwrap(), [1]);

        let operation = <SqliteStore as LogStore<
            Operation<()>,
            VerifyingKey,
            usize,
            SeqNum,
            Hash,
        >>::get_latest_entry(&store, &log_0.author(), &0)
        .await
        .unwrap()
        .unwrap();
        assert_eq!(operation.header.seq_num, 1);

        // Topic "cats" contains one log: 2 with four operations.
        let authors =
            <SqliteStore as TopicStore<[u8; 32], VerifyingKey, usize>>::resolve(&store, &cats)
                .await
                .unwrap();
        assert_eq!(*authors.get(&log_2.author()).unwrap(), [2]);

        let operation = <SqliteStore as LogStore<
            Operation<()>,
            VerifyingKey,
            usize,
            SeqNum,
            Hash,
        >>::get_latest_entry(&store, &log_2.author(), &2)
        .await
        .unwrap()
        .unwrap();
        assert_eq!(operation.header.seq_num, 2);
    }

    #[tokio::test]
    async fn missing_prefix() {
        let store = SqliteStore::temporary().await;
        let signing_key = SigningKey::generate();

        // Create an operation which has already advanced in the log (it has a backlink and higher
        // sequence number).
        let mut header = Header {
            verifying_key: signing_key.verifying_key(),
            version: 1,
            signature: None,
            payload_size: 0,
            payload_hash: None,
            seq_num: 12, // we'll be missing 11 operations between the first and this one
            backlink: Some(Hash::digest(b"mock operation")),
            extensions: (),
        };
        header.sign(&signing_key);

        let operation = Operation {
            hash: header.hash(),
            header,
            body: None,
        };

        let result = ingest_operation(&store, &operation, &1, &1, false).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn ignore_outdated_pruned_operations() {
        let store = SqliteStore::temporary().await;
        let signing_key = SigningKey::generate();

        // 1. Create an advanced operation in a log which assumes that all previous operations have
        //    been pruned.
        let mut header = Header {
            verifying_key: signing_key.verifying_key(),
            version: 1,
            signature: None,
            payload_size: 0,
            payload_hash: None,
            seq_num: 1,
            backlink: Some(Hash::digest(b"mock operation")),
            extensions: (),
        };
        header.sign(&signing_key);

        let operation = Operation {
            hash: header.hash(),
            header,
            body: None,
        };

        let prune_flag = true; // Ingest does not do any pruning, but the flag affects validation.
        let result = ingest_operation(&store, &operation, &1, &1, prune_flag).await;
        assert!(result.is_ok());

        // 2. Create an operation which is from an "outdated" seq from before the log was pruned.
        let mut header = Header {
            verifying_key: signing_key.verifying_key(),
            version: 1,
            signature: None,
            payload_size: 0,
            payload_hash: None,
            seq_num: 0,
            backlink: None,
            extensions: (),
        };
        header.sign(&signing_key);

        let operation = Operation {
            hash: header.hash(),
            header,
            body: None,
        };

        let result = ingest_operation(&store, &operation, &1, &1, false).await;
        assert!(result.is_err());
    }
}