p2panda_stream/spaces/
processor.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use std::borrow::Borrow;
4use std::cell::RefCell;
5use std::collections::VecDeque;
6
7use p2panda_auth::traits::Conditions;
8use p2panda_spaces::Forge;
9use p2panda_spaces::manager::Manager;
10use p2panda_spaces::{AuthMessage, Event, SpacesStoreState};
11use p2panda_store::Transaction;
12use p2panda_store::groups::GroupsStore;
13use p2panda_store::key_registry::KeyRegistryStore;
14use p2panda_store::key_secrets::PreKeyBundlesStore;
15use p2panda_store::spaces::{SpacesMessageStore, SpacesStore};
16use serde::{Deserialize, Serialize};
17use thiserror::Error;
18use tokio::sync::Notify;
19
20use crate::Processor;
21use crate::spaces::SpacesProcessorArgs;
22
23pub type SpacesManager<S, F, C> = Manager<S, F, C>;
24
25#[derive(Clone, Debug, Error)]
26#[error("{0}")]
27pub struct SpacesError(String);
28
29/// Processor for spaces operations.
30pub struct Spaces<T, S, F, C> {
31    manager: SpacesManager<S, F, C>,
32    notify: Notify,
33    queue: RefCell<VecDeque<(T, SpacesResult<C>)>>,
34}
35
36impl<T, S, F, C> Spaces<T, S, F, C> {
37    pub fn new(manager: SpacesManager<S, F, C>) -> Self {
38        Self {
39            manager,
40            notify: Notify::new(),
41            queue: RefCell::new(VecDeque::new()),
42        }
43    }
44}
45
46impl<T, S, F, C> Processor<T> for Spaces<T, S, F, C>
47where
48    T: Borrow<SpacesProcessorArgs<C>>,
49    S: Clone
50        + SpacesStore<SpacesStoreState<C>>
51        + SpacesMessageStore<F::Message>
52        + GroupsStore<AuthMessage<C>, C>
53        + KeyRegistryStore
54        + PreKeyBundlesStore
55        + Transaction,
56    F: Forge<C>,
57    C: Conditions + Serialize + for<'a> Deserialize<'a>,
58{
59    type Output = (T, SpacesResult<C>);
60
61    type Error = (T, SpacesError);
62
63    async fn process(&self, input: T) -> Result<(), Self::Error> {
64        let input_args: &SpacesProcessorArgs<C> = input.borrow();
65
66        let result = match input_args {
67            SpacesProcessorArgs::Process { msg } => match self.manager.process(msg).await {
68                Ok(events) => (input, SpacesResult::Processed { events }),
69                Err(err) => return Err((input, SpacesError(err.to_string()))),
70            },
71            // For locally created operations the spaces args have already been processed and
72            // resulting events are included in the processor args here. All that's needed from the
73            // processor is to forward them onto any consumers.
74            SpacesProcessorArgs::AlreadyProcessed { events, .. } => {
75                let events = events.clone();
76                (input, SpacesResult::Processed { events })
77            }
78            SpacesProcessorArgs::Ignore => (input, SpacesResult::Ignored),
79        };
80
81        self.queue.borrow_mut().push_back(result);
82        self.notify.notify_one();
83
84        Ok(())
85    }
86
87    async fn next(&self) -> Result<Self::Output, Self::Error> {
88        loop {
89            if let Some(item) = self.queue.borrow_mut().pop_front() {
90                return Ok(item);
91            }
92
93            self.notify.notified().await;
94        }
95    }
96}
97
98#[derive(Clone, Debug)]
99pub enum SpacesResult<C> {
100    Processed { events: Vec<Event<C>> },
101    Ignored,
102}
103
104impl<C> SpacesResult<C> {
105    pub fn was_processed(self) -> bool {
106        match self {
107            Self::Processed { .. } => true,
108            Self::Ignored => false,
109        }
110    }
111}
p2panda_stream/spaces/processor.rs

p2panda_stream/spaces/
processor.rs
