p2panda-stream 0.8.0

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

use std::borrow::Borrow;
use std::cell::RefCell;
use std::collections::VecDeque;
use std::marker::PhantomData;

use p2panda_core::{AnyOperation, Hash, LogId, SeqNum, VerifyingKey};
use p2panda_store::Transaction;
use p2panda_store::logs::LogStore;
use thiserror::Error;
use tokio::sync::Notify;

use crate::Processor;
use crate::log_prune::LogPruneArgs;

pub struct LogPrune<S, T, L> {
    store: S,
    notify: Notify,
    queue: RefCell<VecDeque<(T, LogPruneResult)>>,
    _marker: PhantomData<L>,
}

impl<S, T, L> LogPrune<S, T, L>
where
    S: LogStore<AnyOperation, VerifyingKey, L, SeqNum, Hash>,
    L: LogId,
{
    pub fn new(store: S) -> Self {
        Self {
            store,
            notify: Notify::new(),
            queue: RefCell::new(VecDeque::new()),
            _marker: PhantomData,
        }
    }
}

impl<S, T, L> Processor<T> for LogPrune<S, T, L>
where
    S: Transaction + LogStore<AnyOperation, VerifyingKey, L, SeqNum, Hash>,
    T: Borrow<LogPruneArgs<VerifyingKey, L, SeqNum>>,
    L: LogId,
{
    type Output = (T, LogPruneResult);

    type Error = (T, LogPruneError);

    async fn process(&self, input: T) -> Result<(), Self::Error> {
        let args: &LogPruneArgs<VerifyingKey, L, SeqNum> = input.borrow();

        let result = if let LogPruneArgs::PruneEntriesUntil {
            author,
            log_id,
            seq_num,
        } = args
        {
            let permit = match self.store.begin().await {
                Ok(permit) => permit,
                Err(err) => return Err((input, LogPruneError::StoreError(err.to_string()))),
            };

            match self.store.prune_entries_tx(author, log_id, seq_num).await {
                Ok(num_entries) => {
                    if let Err(err) = self.store.commit(permit).await {
                        return Err((input, LogPruneError::StoreError(err.to_string())));
                    };

                    (input, LogPruneResult::Pruned { num_entries })
                }
                Err(err) => {
                    // Return the input arguments next to the error to allow mapping it back to
                    // it's source.
                    return Err((input, LogPruneError::StoreError(err.to_string())));
                }
            }
        } else {
            (input, LogPruneResult::Noop)
        };

        self.queue.borrow_mut().push_back(result);
        self.notify.notify_one(); // Wake up any pending recv.

        Ok(())
    }

    async fn next(&self) -> Result<Self::Output, Self::Error> {
        loop {
            if let Some(item) = self.queue.borrow_mut().pop_front() {
                return Ok(item);
            }

            // Wait for notification that an item was added.
            self.notify.notified().await;
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LogPruneResult {
    Noop,
    Pruned { num_entries: u64 },
}

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