fedimint_client_module/
oplog.rs

1use std::fmt::Debug;
2use std::future;
3use std::time::SystemTime;
4
5use fedimint_core::core::OperationId;
6use fedimint_core::db::{Database, DatabaseTransaction};
7use fedimint_core::encoding::{Decodable, DecodeError, Encodable};
8use fedimint_core::module::registry::ModuleDecoderRegistry;
9use fedimint_core::task::{MaybeSend, MaybeSync};
10use fedimint_core::util::BoxStream;
11use fedimint_core::{apply, async_trait_maybe_send};
12use futures::{StreamExt, stream};
13use serde::de::DeserializeOwned;
14use serde::{Deserialize, Serialize};
15
16/// Json value using string representation as db encoding.
17#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
18#[serde(transparent)]
19pub struct JsonStringed(pub serde_json::Value);
20
21#[cfg(feature = "uniffi")]
22uniffi::custom_type!(JsonStringed, String, {
23    lower: |j| serde_json::to_string(&j.0).expect("JSON serialization should not fail"),
24    try_lift: |s| {
25        let value = serde_json::from_str(&s).map_err(|e| DecodeError::from_err(e))?;
26        Ok(JsonStringed(value))
27    },
28});
29
30impl Encodable for JsonStringed {
31    fn consensus_encode<W: std::io::Write>(&self, writer: &mut W) -> Result<(), std::io::Error> {
32        let json_str = serde_json::to_string(&self.0).expect("JSON serialization should not fail");
33        json_str.consensus_encode(writer)
34    }
35}
36
37impl Decodable for JsonStringed {
38    fn consensus_decode_partial<R: std::io::Read>(
39        r: &mut R,
40        modules: &ModuleDecoderRegistry,
41    ) -> Result<Self, DecodeError> {
42        let json_str = String::consensus_decode_partial(r, modules)?;
43        let value = serde_json::from_str(&json_str).map_err(DecodeError::from_err)?;
44        Ok(JsonStringed(value))
45    }
46}
47
48#[apply(async_trait_maybe_send!)]
49pub trait IOperationLog {
50    async fn get_operation(&self, operation_id: OperationId) -> Option<OperationLogEntry>;
51
52    async fn get_operation_dbtx(
53        &self,
54        dbtx: &mut DatabaseTransaction<'_>,
55        operation_id: OperationId,
56    ) -> Option<OperationLogEntry>;
57
58    async fn add_operation_log_entry_dbtx(
59        &self,
60        dbtx: &mut DatabaseTransaction<'_>,
61        operation_id: OperationId,
62        operation_type: &str,
63        operation_meta: serde_json::Value,
64    );
65
66    fn outcome_or_updates(
67        &self,
68        db: &Database,
69        operation_id: OperationId,
70        operation_log_entry: OperationLogEntry,
71        stream_gen: Box<dyn FnOnce() -> BoxStream<'static, serde_json::Value>>,
72    ) -> UpdateStreamOrOutcome<serde_json::Value>;
73}
74
75/// Represents the outcome of an operation, combining both the outcome value and
76/// its timestamp
77#[derive(Debug, Clone, Serialize, Deserialize, Encodable, Decodable, PartialEq, Eq)]
78#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
79pub struct OperationOutcome {
80    pub time: SystemTime,
81    pub outcome: JsonStringed,
82}
83
84/// Represents an operation triggered by a user, typically related to sending or
85/// receiving money.
86///
87/// There are three levels of introspection possible for `OperationLogEntry`s:
88///   1. The [`OperationLogEntry::operation_module_kind`] function returns the
89///      kind of the module that created the operation.
90///   2. The [`OperationLogEntry::meta`] function returns static meta data that
91///      was associated with the operation when it was created. Modules define
92///      their own meta structures, so the module kind has to be used to
93///      determine the structure of the meta data.
94///   3. To find out the current state of the operation there is a two-step
95///      process:
96///      * First, the [`OperationLogEntry::outcome`] function returns the
97///        outcome if the operation finished **and** the update subscription
98///        stream has been processed till its end at least once.
99///      * If that isn't the case, the [`OperationLogEntry::outcome`] method
100///        will return `None` and the appropriate update subscription function
101///        has to be called. See the respective client extension trait for these
102///        functions.
103#[derive(Debug, Serialize, Deserialize, Encodable, Decodable)]
104#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
105pub struct OperationLogEntry {
106    pub(crate) operation_module_kind: String,
107    pub(crate) meta: JsonStringed,
108    // TODO: probably change all that JSON to Dyn-types
109    pub(crate) outcome: Option<OperationOutcome>,
110}
111
112impl OperationLogEntry {
113    pub fn new(
114        operation_module_kind: String,
115        meta: JsonStringed,
116        outcome: Option<OperationOutcome>,
117    ) -> Self {
118        Self {
119            operation_module_kind,
120            meta,
121            outcome,
122        }
123    }
124
125    /// Returns the kind of the module that generated the operation
126    pub fn operation_module_kind(&self) -> &str {
127        &self.operation_module_kind
128    }
129
130    /// Returns the meta data of the operation. This is a JSON value that can be
131    /// either returned as a [`serde_json::Value`] or deserialized into a
132    /// specific type. The specific type should be named `<Module>OperationMeta`
133    /// in the module's client crate. The module can be determined by calling
134    /// [`OperationLogEntry::operation_module_kind`].
135    pub fn meta<M: DeserializeOwned>(&self) -> M {
136        self.try_meta()
137            .expect("JSON deserialization should not fail")
138    }
139
140    /// Fallible version of [`OperationLogEntry::meta`]. Used to avoid panics in
141    /// the case of failed past migrations, resulting in invalid encodings in
142    /// the DB.
143    pub fn try_meta<M: DeserializeOwned>(&self) -> Result<M, serde_json::Error> {
144        serde_json::from_value(self.meta.0.clone())
145    }
146
147    /// Returns the last state update of the operation, if any was cached yet.
148    /// If this hasn't been the case yet and `None` is returned subscribe to the
149    /// appropriate update stream.
150    ///
151    /// ## Determining the return type
152    /// [`OperationLogEntry::meta`] should tell you the which operation type of
153    /// a given module the outcome belongs to. The operation type will have a
154    /// corresponding `async fn subscribe_type(&self, operation_id:
155    /// OperationId) -> anyhow::Result<UpdateStreamOrOutcome<TypeState>>;`
156    /// function that returns a `UpdateStreamOrOutcome<S>` where `S` is the
157    /// high-level state the operation is in. If this state is terminal, i.e.
158    /// the stream closes after returning it, it will be cached as the `outcome`
159    /// of the operation.
160    ///
161    /// This means the type to be used for deserializing the outcome is `S`,
162    /// often called `<OperationType>State`. Alternatively one can also use
163    /// [`serde_json::Value`] to get the unstructured data.
164    pub fn outcome<D: DeserializeOwned>(&self) -> Option<D> {
165        self.try_outcome()
166            .expect("JSON deserialization should not fail")
167    }
168
169    /// Fallible version of [`OperationLogEntry::outcome`]. Used to avoid panics
170    /// in the case of failed past migrations, resulting in invalid encodings in
171    /// the DB.
172    pub fn try_outcome<D: DeserializeOwned>(&self) -> Result<Option<D>, serde_json::Error> {
173        self.outcome
174            .as_ref()
175            .map(|outcome| serde_json::from_value(outcome.outcome.0.clone()))
176            .transpose()
177    }
178
179    /// Returns the time when the outcome was cached.
180    pub fn outcome_time(&self) -> Option<SystemTime> {
181        self.outcome.as_ref().map(|o| o.time)
182    }
183
184    pub fn set_outcome(&mut self, outcome: impl Into<Option<OperationOutcome>>) {
185        self.outcome = outcome.into();
186    }
187}
188
189/// Either a stream of operation updates if the operation hasn't finished yet or
190/// its outcome otherwise.
191pub enum UpdateStreamOrOutcome<U> {
192    UpdateStream(BoxStream<'static, U>),
193    Outcome(U),
194}
195
196impl<U: Debug> Debug for UpdateStreamOrOutcome<U> {
197    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
198        match self {
199            UpdateStreamOrOutcome::UpdateStream(_) => write!(f, "UpdateStream"),
200            UpdateStreamOrOutcome::Outcome(o) => f.debug_tuple("Outcome").field(o).finish(),
201        }
202    }
203}
204
205impl<U> UpdateStreamOrOutcome<U>
206where
207    U: MaybeSend + MaybeSync + 'static,
208{
209    /// Returns a stream no matter if the operation is finished. If there
210    /// already is a cached outcome the stream will only return that, otherwise
211    /// all updates will be returned until the operation finishes.
212    pub fn into_stream(self) -> BoxStream<'static, U> {
213        match self {
214            UpdateStreamOrOutcome::UpdateStream(stream) => stream,
215            UpdateStreamOrOutcome::Outcome(outcome) => {
216                Box::pin(stream::once(future::ready(outcome)))
217            }
218        }
219    }
220
221    /// Awaits the outcome of the operation update stream, either by returning
222    /// the cached value or by consuming the entire stream and returning the
223    /// last update.
224    pub async fn await_outcome(self) -> Option<U> {
225        match self {
226            UpdateStreamOrOutcome::Outcome(outcome) => Some(outcome),
227            UpdateStreamOrOutcome::UpdateStream(mut stream) => {
228                let mut last_update = None;
229                while let Some(update) = stream.next().await {
230                    last_update = Some(update);
231                }
232                last_update
233            }
234        }
235    }
236}
237
238#[cfg(test)]
239mod tests {
240    use futures::stream;
241    use serde_json::Value;
242
243    use super::*;
244
245    #[tokio::test]
246    async fn test_await_outcome_cached() {
247        let test_value = serde_json::json!({"status": "completed", "amount": 100});
248        let cached_outcome = UpdateStreamOrOutcome::Outcome(test_value.clone());
249        let result = cached_outcome.await_outcome().await;
250        assert_eq!(result, Some(test_value));
251    }
252
253    #[tokio::test]
254    async fn test_await_outcome_uncached_with_updates() {
255        let update_stream = Box::pin(stream::iter(vec![
256            Value::from(0),
257            Value::from(1),
258            Value::from(2),
259        ]));
260        let uncached_outcome = UpdateStreamOrOutcome::UpdateStream(update_stream);
261        let result = uncached_outcome.await_outcome().await;
262        assert_eq!(result, Some(Value::from(2)));
263    }
264
265    #[tokio::test]
266    async fn test_await_outcome_uncached_empty_stream() {
267        let empty_stream = Box::pin(stream::empty::<serde_json::Value>());
268        let uncached_outcome = UpdateStreamOrOutcome::UpdateStream(empty_stream);
269        let result = uncached_outcome.await_outcome().await;
270        assert_eq!(result, None);
271    }
272
273    #[tokio::test]
274    async fn test_await_outcome_uncached_single_update() {
275        let update_stream = Box::pin(stream::once(async { Value::from(0) }));
276        let uncached_outcome = UpdateStreamOrOutcome::UpdateStream(update_stream);
277        let result = uncached_outcome.await_outcome().await;
278        assert_eq!(result, Some(Value::from(0)));
279    }
280}
fedimint_client_module/oplog.rs

fedimint_client_module/
oplog.rs
