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::manager::{Manager, ManagerError};
9use p2panda_spaces::{AuthMessage, Event, SpacesStoreState, StrongRemoveResolver};
10use p2panda_spaces::{Forge, SpacesArgs};
11use p2panda_store::Transaction;
12use p2panda_store::groups::GroupsStore;
13use p2panda_store::key_registry::KeyRegistryStore;
14use p2panda_store::key_secrets::KeySecretsStore;
15use p2panda_store::spaces::{SpacesMessageStore, SpacesStore};
16use tokio::sync::Notify;
17use tracing::trace;
18
19use crate::Processor;
20use crate::spaces::SpacesProcessorArgs;
21
22pub type SpacesManager<S, F, C> = Manager<S, F, C, StrongRemoveResolver<C>>;
23
24pub type SpacesManagerError<F, C> = ManagerError<F, C, StrongRemoveResolver<C>>;
25
26#[derive(Clone)]
27pub enum SpacesResult<C> {
28    Processed { events: Vec<Event<C>> },
29    Ignored,
30}
31
32impl<C> SpacesResult<C> {
33    pub fn was_processed(self) -> bool {
34        match self {
35            Self::Processed { .. } => true,
36            Self::Ignored => false,
37        }
38    }
39}
40
41/// Processor for spaces operations.
42pub struct Spaces<T, S, F, C> {
43    manager: SpacesManager<S, F, C>,
44    notify: Notify,
45    queue: RefCell<VecDeque<(T, SpacesResult<C>)>>,
46}
47
48impl<T, S, F, C> Spaces<T, S, F, C> {
49    pub fn new(manager: SpacesManager<S, F, C>) -> Self {
50        Self {
51            manager,
52            notify: Notify::new(),
53            queue: RefCell::new(VecDeque::new()),
54        }
55    }
56}
57
58impl<T, S, F, C> Processor<T> for Spaces<T, S, F, C>
59where
60    T: Borrow<SpacesProcessorArgs<C>>,
61    S: Clone
62        + Transaction
63        + SpacesStore<SpacesStoreState<C>>
64        + SpacesMessageStore<SpacesArgs<C>>
65        + GroupsStore<AuthMessage<C>, C>
66        + KeyRegistryStore
67        + KeySecretsStore,
68    F: Forge<C>,
69    C: Conditions,
70{
71    type Output = (T, SpacesResult<C>);
72
73    type Error = (T, SpacesManagerError<F, C>);
74
75    async fn process(&self, input: T) -> Result<(), Self::Error> {
76        let input_args: &SpacesProcessorArgs<C> = input.borrow();
77
78        let result = if let SpacesProcessorArgs::Process { msg } = input_args {
79            let (groups_y, space_y, events) = match self.manager.process(msg).await {
80                Ok(result) => result,
81                Err(err) => return Err((input, err)),
82            };
83
84            if let Some(_groups_y) = groups_y {
85                // @TODO: persist groups state.
86            }
87
88            if let Some(_space_y) = space_y {
89                // @TODO: persist space state.
90            }
91
92            (input, SpacesResult::Processed { events })
93        } else {
94            trace!("ignore non-spaces operation");
95            (input, SpacesResult::Ignored)
96        };
97
98        self.queue.borrow_mut().push_back(result);
99        self.notify.notify_one();
100
101        Ok(())
102    }
103
104    async fn next(&self) -> Result<Self::Output, Self::Error> {
105        loop {
106            if let Some(item) = self.queue.borrow_mut().pop_front() {
107                return Ok(item);
108            }
109
110            self.notify.notified().await;
111        }
112    }
113}
p2panda_stream/spaces/processor.rs

p2panda_stream/spaces/
processor.rs
