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

fedimint_client_module/
oplog.rs
