fedimint_client_module/module/
mod.rs

1use core::fmt;
2use std::any::Any;
3use std::collections::BTreeSet;
4use std::fmt::Debug;
5use std::pin::Pin;
6use std::sync::{Arc, Weak};
7use std::{ffi, marker, ops};
8
9use anyhow::{anyhow, bail};
10use bitcoin::secp256k1::PublicKey;
11use fedimint_api_client::api::{DynGlobalApi, DynModuleApi};
12use fedimint_core::config::ClientConfig;
13use fedimint_core::core::{
14    Decoder, DynInput, DynOutput, IInput, IntoDynInstance, ModuleInstanceId, ModuleKind,
15    OperationId,
16};
17use fedimint_core::db::{Database, DatabaseTransaction, GlobalDBTxAccessToken, NonCommittable};
18use fedimint_core::invite_code::InviteCode;
19use fedimint_core::module::registry::{ModuleDecoderRegistry, ModuleRegistry};
20use fedimint_core::module::{AmountUnit, Amounts, CommonModuleInit, ModuleCommon, ModuleInit};
21use fedimint_core::task::{MaybeSend, MaybeSync};
22use fedimint_core::util::{BoxStream, FmtCompact as _};
23use fedimint_core::{
24    Amount, OutPoint, PeerId, apply, async_trait_maybe_send, dyn_newtype_define, maybe_add_send,
25    maybe_add_send_sync,
26};
27use fedimint_eventlog::{
28    DBTransactionEventLogExt, Event, EventKind, EventLogId, EventPersistence, PersistedLogEntry,
29};
30use fedimint_logging::LOG_CLIENT;
31use futures::{Stream, StreamExt};
32use serde::Serialize;
33use serde::de::DeserializeOwned;
34use tracing::warn;
35
36use self::init::ClientModuleInit;
37use crate::module::recovery::{DynModuleBackup, ModuleBackup};
38use crate::oplog::{IOperationLog, OperationLogEntry, UpdateStreamOrOutcome};
39use crate::sm::executor::{ActiveStateKey, IExecutor, InactiveStateKey};
40use crate::sm::{self, ActiveStateMeta, Context, DynContext, DynState, InactiveStateMeta, State};
41use crate::transaction::{
42    ClientInputBundle, ClientOutputBundle, FeeQuote, FeeQuoteRequest, TransactionBuilder,
43};
44use crate::{AddStateMachinesResult, InstancelessDynClientInputBundle, TransactionUpdates, oplog};
45
46pub mod init;
47pub mod recovery;
48
49pub type ClientModuleRegistry = ModuleRegistry<DynClientModule>;
50
51/// A fedimint-client interface exposed to client modules
52///
53/// To break the dependency of the client modules on the whole fedimint client
54/// and in particular the `fedimint-client` crate, the module gets access to an
55/// interface, that is implemented by the `Client`.
56///
57/// This allows lose coupling, less recompilation and better control and
58/// understanding of what functionality of the Client the modules get access to.
59#[apply(async_trait_maybe_send!)]
60pub trait ClientContextIface: MaybeSend + MaybeSync {
61    fn get_module(&self, instance: ModuleInstanceId) -> &maybe_add_send_sync!(dyn IClientModule);
62    fn api_clone(&self) -> DynGlobalApi;
63    fn decoders(&self) -> &ModuleDecoderRegistry;
64    async fn finalize_and_submit_transaction(
65        &self,
66        operation_id: OperationId,
67        operation_type: &str,
68        operation_meta_gen: Box<maybe_add_send_sync!(dyn Fn(OutPointRange) -> serde_json::Value)>,
69        tx_builder: TransactionBuilder,
70    ) -> anyhow::Result<OutPointRange>;
71
72    async fn finalize_and_submit_transaction_dbtx(
73        &self,
74        dbtx: &mut DatabaseTransaction<'_>,
75        operation_id: OperationId,
76        operation_type: &str,
77        operation_meta_gen: Box<maybe_add_send_sync!(dyn Fn(OutPointRange) -> serde_json::Value)>,
78        tx_builder: TransactionBuilder,
79    ) -> anyhow::Result<OutPointRange>;
80
81    // TODO: unify
82    async fn finalize_and_submit_transaction_inner(
83        &self,
84        dbtx: &mut DatabaseTransaction<'_>,
85        operation_id: OperationId,
86        tx_builder: TransactionBuilder,
87    ) -> anyhow::Result<OutPointRange>;
88
89    /// Computes the fee finalizing and submitting a transaction with the
90    /// explicit items described by `request` would incur, without submitting
91    /// anything. See `Client::fee_quote`.
92    async fn fee_quote(
93        &self,
94        operation_id: OperationId,
95        request: FeeQuoteRequest,
96    ) -> anyhow::Result<FeeQuote>;
97
98    /// The client's balance for `unit`, held by the primary module. See
99    /// `Client::get_balance_for_unit`.
100    async fn get_balance_for_unit(&self, unit: AmountUnit) -> anyhow::Result<Amount>;
101
102    async fn transaction_updates(&self, operation_id: OperationId) -> TransactionUpdates;
103
104    async fn await_primary_module_outputs(
105        &self,
106        operation_id: OperationId,
107        // TODO: make `impl Iterator<Item = ...>`
108        outputs: Vec<OutPoint>,
109    ) -> anyhow::Result<()>;
110
111    fn operation_log(&self) -> &dyn IOperationLog;
112
113    async fn has_active_states(&self, operation_id: OperationId) -> bool;
114
115    async fn operation_exists(&self, operation_id: OperationId) -> bool;
116
117    async fn config(&self) -> ClientConfig;
118
119    fn db(&self) -> &Database;
120
121    fn executor(&self) -> &(maybe_add_send_sync!(dyn IExecutor + 'static));
122
123    async fn invite_code(&self, peer: PeerId) -> Option<InviteCode>;
124
125    fn get_internal_payment_markers(&self) -> anyhow::Result<(PublicKey, u64)>;
126
127    #[allow(clippy::too_many_arguments)]
128    async fn log_event_json(
129        &self,
130        dbtx: &mut DatabaseTransaction<'_, NonCommittable>,
131        module_kind: Option<ModuleKind>,
132        module_id: ModuleInstanceId,
133        kind: EventKind,
134        payload: serde_json::Value,
135        persist: EventPersistence,
136    );
137
138    async fn read_operation_active_states<'dbtx>(
139        &self,
140        operation_id: OperationId,
141        module_id: ModuleInstanceId,
142        dbtx: &'dbtx mut DatabaseTransaction<'_>,
143    ) -> Pin<Box<maybe_add_send!(dyn Stream<Item = (ActiveStateKey, ActiveStateMeta)> + 'dbtx)>>;
144
145    async fn read_operation_inactive_states<'dbtx>(
146        &self,
147        operation_id: OperationId,
148        module_id: ModuleInstanceId,
149        dbtx: &'dbtx mut DatabaseTransaction<'_>,
150    ) -> Pin<Box<maybe_add_send!(dyn Stream<Item = (InactiveStateKey, InactiveStateMeta)> + 'dbtx)>>;
151}
152
153/// A final, fully initialized client
154///
155/// Client modules need to be able to access a `Client` they are a part
156/// of. To break the circular dependency, the final `Client` is passed
157/// after `Client` was built via a shared state.
158#[derive(Clone, Default)]
159pub struct FinalClientIface(Arc<std::sync::OnceLock<Weak<dyn ClientContextIface>>>);
160
161impl FinalClientIface {
162    /// Get a temporary strong reference to [`ClientContextIface`]
163    ///
164    /// Care must be taken to not let the user take ownership of this value,
165    /// and not store it elsewhere permanently either, as it could prevent
166    /// the cleanup of the Client.
167    pub(crate) fn get(&self) -> Arc<dyn ClientContextIface> {
168        self.0
169            .get()
170            .expect("client must be already set")
171            .upgrade()
172            .expect("client module context must not be use past client shutdown")
173    }
174
175    pub fn set(&self, client: Weak<dyn ClientContextIface>) {
176        self.0.set(client).expect("FinalLazyClient already set");
177    }
178}
179
180impl fmt::Debug for FinalClientIface {
181    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
182        f.write_str("FinalClientIface")
183    }
184}
185/// A Client context for a [`ClientModule`] `M`
186///
187/// Client modules can interact with the whole
188/// client through this struct.
189pub struct ClientContext<M> {
190    client: FinalClientIface,
191    module_instance_id: ModuleInstanceId,
192    global_dbtx_access_token: GlobalDBTxAccessToken,
193    module_db: Database,
194    _marker: marker::PhantomData<M>,
195}
196
197impl<M> Clone for ClientContext<M> {
198    fn clone(&self) -> Self {
199        Self {
200            client: self.client.clone(),
201            module_db: self.module_db.clone(),
202            module_instance_id: self.module_instance_id,
203            _marker: marker::PhantomData,
204            global_dbtx_access_token: self.global_dbtx_access_token,
205        }
206    }
207}
208
209/// A reference back to itself that the module cacn get from the
210/// [`ClientContext`]
211pub struct ClientContextSelfRef<'s, M> {
212    // we are OK storing `ClientStrong` here, because of the `'s` preventing `Self` from being
213    // stored permanently somewhere
214    client: Arc<dyn ClientContextIface>,
215    module_instance_id: ModuleInstanceId,
216    _marker: marker::PhantomData<&'s M>,
217}
218
219impl<M> ops::Deref for ClientContextSelfRef<'_, M>
220where
221    M: ClientModule,
222{
223    type Target = M;
224
225    fn deref(&self) -> &Self::Target {
226        self.client
227            .get_module(self.module_instance_id)
228            .as_any()
229            .downcast_ref::<M>()
230            .unwrap_or_else(|| panic!("Module is not of type {}", std::any::type_name::<M>()))
231    }
232}
233
234impl<M> fmt::Debug for ClientContext<M> {
235    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
236        f.write_str("ClientContext")
237    }
238}
239
240impl<M> ClientContext<M>
241where
242    M: ClientModule,
243{
244    pub fn new(
245        client: FinalClientIface,
246        module_instance_id: ModuleInstanceId,
247        global_dbtx_access_token: GlobalDBTxAccessToken,
248        module_db: Database,
249    ) -> Self {
250        Self {
251            client,
252            module_instance_id,
253            global_dbtx_access_token,
254            module_db,
255            _marker: marker::PhantomData,
256        }
257    }
258
259    /// Get a reference back to client module from the [`Self`]
260    ///
261    /// It's often necessary for a client module to "move self"
262    /// by-value, especially due to async lifetimes issues.
263    /// Clients usually work with `&mut self`, which can't really
264    /// work in such context.
265    ///
266    /// Fortunately [`ClientContext`] is `Clone` and `Send, and
267    /// can be used to recover the reference to the module at later
268    /// time.
269    #[allow(clippy::needless_lifetimes)] // just for explicitiness
270    pub fn self_ref(&self) -> ClientContextSelfRef<'_, M> {
271        ClientContextSelfRef {
272            client: self.client.get(),
273            module_instance_id: self.module_instance_id,
274            _marker: marker::PhantomData,
275        }
276    }
277
278    /// Get a reference to a global Api handle
279    pub fn global_api(&self) -> DynGlobalApi {
280        self.client.get().api_clone()
281    }
282
283    /// Get a reference to a module Api handle
284    pub fn module_api(&self) -> DynModuleApi {
285        self.global_api().with_module(self.module_instance_id)
286    }
287
288    /// A set of all decoders of all modules of the client
289    pub fn decoders(&self) -> ModuleDecoderRegistry {
290        Clone::clone(self.client.get().decoders())
291    }
292
293    pub fn input_from_dyn<'i>(
294        &self,
295        input: &'i DynInput,
296    ) -> Option<&'i <M::Common as ModuleCommon>::Input> {
297        (input.module_instance_id() == self.module_instance_id).then(|| {
298            input
299                .as_any()
300                .downcast_ref::<<M::Common as ModuleCommon>::Input>()
301                .unwrap_or_else(|| {
302                    panic!("instance_id {} just checked", input.module_instance_id())
303                })
304        })
305    }
306
307    pub fn output_from_dyn<'o>(
308        &self,
309        output: &'o DynOutput,
310    ) -> Option<&'o <M::Common as ModuleCommon>::Output> {
311        (output.module_instance_id() == self.module_instance_id).then(|| {
312            output
313                .as_any()
314                .downcast_ref::<<M::Common as ModuleCommon>::Output>()
315                .unwrap_or_else(|| {
316                    panic!("instance_id {} just checked", output.module_instance_id())
317                })
318        })
319    }
320
321    pub fn map_dyn<'s, 'i, 'o, I>(
322        &'s self,
323        typed: impl IntoIterator<Item = I> + 'i,
324    ) -> impl Iterator<Item = <I as IntoDynInstance>::DynType> + 'o
325    where
326        I: IntoDynInstance,
327        'i: 'o,
328        's: 'o,
329    {
330        typed.into_iter().map(|i| self.make_dyn(i))
331    }
332
333    /// Turn a typed output into a dyn version
334    pub fn make_dyn_output(&self, output: <M::Common as ModuleCommon>::Output) -> DynOutput {
335        self.make_dyn(output)
336    }
337
338    /// Turn a typed input into a dyn version
339    pub fn make_dyn_input(&self, input: <M::Common as ModuleCommon>::Input) -> DynInput {
340        self.make_dyn(input)
341    }
342
343    /// Turn a `typed` into a dyn version
344    pub fn make_dyn<I>(&self, typed: I) -> <I as IntoDynInstance>::DynType
345    where
346        I: IntoDynInstance,
347    {
348        typed.into_dyn(self.module_instance_id)
349    }
350
351    /// Turn a typed [`ClientOutputBundle`] into a dyn version
352    pub fn make_client_outputs<O, S>(&self, output: ClientOutputBundle<O, S>) -> ClientOutputBundle
353    where
354        O: IntoDynInstance<DynType = DynOutput> + 'static,
355        S: IntoDynInstance<DynType = DynState> + 'static,
356    {
357        self.make_dyn(output)
358    }
359
360    /// Turn a typed [`ClientInputBundle`] into a dyn version
361    pub fn make_client_inputs<I, S>(&self, inputs: ClientInputBundle<I, S>) -> ClientInputBundle
362    where
363        I: IntoDynInstance<DynType = DynInput> + 'static,
364        S: IntoDynInstance<DynType = DynState> + 'static,
365    {
366        self.make_dyn(inputs)
367    }
368
369    pub fn make_dyn_state<S>(&self, sm: S) -> DynState
370    where
371        S: sm::IState + 'static,
372    {
373        DynState::from_typed(self.module_instance_id, sm)
374    }
375
376    pub async fn finalize_and_submit_transaction<F, Meta>(
377        &self,
378        operation_id: OperationId,
379        operation_type: &str,
380        operation_meta_gen: F,
381        tx_builder: TransactionBuilder,
382    ) -> anyhow::Result<OutPointRange>
383    where
384        F: Fn(OutPointRange) -> Meta + Clone + MaybeSend + MaybeSync + 'static,
385        Meta: serde::Serialize + MaybeSend,
386    {
387        self.client
388            .get()
389            .finalize_and_submit_transaction(
390                operation_id,
391                operation_type,
392                Box::new(move |out_point_range| {
393                    serde_json::to_value(operation_meta_gen(out_point_range)).expect("Can't fail")
394                }),
395                tx_builder,
396            )
397            .await
398    }
399
400    pub async fn finalize_and_submit_transaction_dbtx<F, Meta>(
401        &self,
402        dbtx: &mut DatabaseTransaction<'_>,
403        operation_id: OperationId,
404        operation_type: &str,
405        operation_meta_gen: F,
406        tx_builder: TransactionBuilder,
407    ) -> anyhow::Result<OutPointRange>
408    where
409        F: Fn(OutPointRange) -> Meta + MaybeSend + MaybeSync + 'static,
410        Meta: serde::Serialize + MaybeSend,
411    {
412        self.client
413            .get()
414            .finalize_and_submit_transaction_dbtx(
415                &mut dbtx.global_dbtx(self.global_dbtx_access_token),
416                operation_id,
417                operation_type,
418                Box::new(move |out_point_range| {
419                    serde_json::to_value(operation_meta_gen(out_point_range)).expect("Can't fail")
420                }),
421                tx_builder,
422            )
423            .await
424    }
425
426    /// Computes the fee that finalizing and submitting a transaction with the
427    /// explicit items described by `request` would incur, as a dry-run over the
428    /// client's current funds, without submitting anything. See
429    /// `Client::fee_quote`.
430    ///
431    /// Summarize the explicit inputs/outputs the operation would submit (their
432    /// gross amounts and federation fees) in `request`; the returned breakdown
433    /// adds the change the primary module's balancing would produce.
434    pub async fn fee_quote(
435        &self,
436        operation_id: OperationId,
437        request: FeeQuoteRequest,
438    ) -> anyhow::Result<FeeQuote> {
439        self.client.get().fee_quote(operation_id, request).await
440    }
441
442    /// The client's Bitcoin balance, held by the primary module. See
443    /// `Client::get_balance_for_btc`.
444    pub async fn get_balance_for_btc(&self) -> anyhow::Result<Amount> {
445        self.client
446            .get()
447            .get_balance_for_unit(AmountUnit::BITCOIN)
448            .await
449    }
450
451    pub async fn transaction_updates(&self, operation_id: OperationId) -> TransactionUpdates {
452        self.client.get().transaction_updates(operation_id).await
453    }
454
455    pub async fn await_primary_module_outputs(
456        &self,
457        operation_id: OperationId,
458        // TODO: make `impl Iterator<Item = ...>`
459        outputs: Vec<OutPoint>,
460    ) -> anyhow::Result<()> {
461        self.client
462            .get()
463            .await_primary_module_outputs(operation_id, outputs)
464            .await
465    }
466
467    // TODO: unify with `Self::get_operation`
468    pub async fn get_operation(
469        &self,
470        operation_id: OperationId,
471    ) -> anyhow::Result<oplog::OperationLogEntry> {
472        let operation = self
473            .client
474            .get()
475            .operation_log()
476            .get_operation(operation_id)
477            .await
478            .ok_or(anyhow::anyhow!("Operation not found"))?;
479
480        if operation.operation_module_kind() != M::kind().as_str() {
481            bail!("Operation is not a lightning operation");
482        }
483
484        Ok(operation)
485    }
486
487    /// Get global db.
488    ///
489    /// Only intended for internal use (private).
490    fn global_db(&self) -> fedimint_core::db::Database {
491        let db = Clone::clone(self.client.get().db());
492
493        db.ensure_global()
494            .expect("global_db must always return a global db");
495
496        db
497    }
498
499    pub fn module_db(&self) -> &Database {
500        self.module_db
501            .ensure_isolated()
502            .expect("module_db must always return isolated db");
503        &self.module_db
504    }
505
506    /// Read a portion of the client's event log, starting at `pos` (or the
507    /// beginning of the log if `None`) and returning up to `limit` entries.
508    pub async fn get_event_log(
509        &self,
510        pos: Option<EventLogId>,
511        limit: u64,
512    ) -> Vec<PersistedLogEntry> {
513        self.global_db()
514            .begin_transaction_nc()
515            .await
516            .get_event_log(pos, limit)
517            .await
518    }
519
520    pub async fn has_active_states(&self, op_id: OperationId) -> bool {
521        self.client.get().has_active_states(op_id).await
522    }
523
524    pub async fn operation_exists(&self, op_id: OperationId) -> bool {
525        self.client.get().operation_exists(op_id).await
526    }
527
528    pub async fn get_own_active_states(&self) -> Vec<(M::States, ActiveStateMeta)> {
529        self.client
530            .get()
531            .executor()
532            .get_active_states()
533            .await
534            .into_iter()
535            .filter(|s| s.0.module_instance_id() == self.module_instance_id)
536            .map(|s| {
537                (
538                    Clone::clone(
539                        s.0.as_any()
540                            .downcast_ref::<M::States>()
541                            .expect("incorrect output type passed to module plugin"),
542                    ),
543                    s.1,
544                )
545            })
546            .collect()
547    }
548
549    /// Returns this module's currently active state machines for the operation.
550    pub async fn get_own_operation_active_states(
551        &self,
552        operation_id: OperationId,
553    ) -> Vec<(M::States, ActiveStateMeta)> {
554        let db = self.global_db();
555        let mut dbtx = db.begin_transaction_nc().await;
556
557        self.client
558            .get()
559            .read_operation_active_states(operation_id, self.module_instance_id, &mut dbtx)
560            .await
561            .map(|(key, meta)| {
562                (
563                    Clone::clone(
564                        key.state
565                            .as_any()
566                            .downcast_ref::<M::States>()
567                            .expect("incorrect output type passed to module plugin"),
568                    ),
569                    meta,
570                )
571            })
572            .collect()
573            .await
574    }
575
576    /// Returns this module's previously active state machines for the
577    /// operation.
578    pub async fn get_own_operation_inactive_states(
579        &self,
580        operation_id: OperationId,
581    ) -> Vec<(M::States, InactiveStateMeta)> {
582        let db = self.global_db();
583        let mut dbtx = db.begin_transaction_nc().await;
584
585        self.client
586            .get()
587            .read_operation_inactive_states(operation_id, self.module_instance_id, &mut dbtx)
588            .await
589            .map(|(key, meta)| {
590                (
591                    Clone::clone(
592                        key.state
593                            .as_any()
594                            .downcast_ref::<M::States>()
595                            .expect("incorrect output type passed to module plugin"),
596                    ),
597                    meta,
598                )
599            })
600            .collect()
601            .await
602    }
603
604    pub async fn get_config(&self) -> ClientConfig {
605        self.client.get().config().await
606    }
607
608    /// Returns an invite code for the federation that points to an arbitrary
609    /// guardian server for fetching the config
610    pub async fn get_invite_code(&self) -> InviteCode {
611        let cfg = self.get_config().await.global;
612        self.client
613            .get()
614            .invite_code(
615                *cfg.api_endpoints
616                    .keys()
617                    .next()
618                    .expect("A federation always has at least one guardian"),
619            )
620            .await
621            .expect("The guardian we requested an invite code for exists")
622    }
623
624    pub fn get_internal_payment_markers(&self) -> anyhow::Result<(PublicKey, u64)> {
625        self.client.get().get_internal_payment_markers()
626    }
627
628    /// This method starts n state machines with given operation id without a
629    /// corresponding transaction
630    pub async fn manual_operation_start(
631        &self,
632        operation_id: OperationId,
633        op_type: &str,
634        operation_meta: impl serde::Serialize + Debug,
635        sms: Vec<DynState>,
636    ) -> anyhow::Result<()> {
637        let db = self.module_db();
638        let mut dbtx = db.begin_transaction().await;
639        {
640            let dbtx = &mut dbtx.global_dbtx(self.global_dbtx_access_token);
641
642            self.manual_operation_start_inner(
643                &mut dbtx.to_ref_nc(),
644                operation_id,
645                op_type,
646                operation_meta,
647                sms,
648            )
649            .await?;
650        }
651
652        dbtx.commit_tx_result().await.map_err(|_| {
653            anyhow!(
654                "Operation with id {} already exists",
655                operation_id.fmt_short()
656            )
657        })?;
658
659        Ok(())
660    }
661
662    pub async fn manual_operation_start_dbtx(
663        &self,
664        dbtx: &mut DatabaseTransaction<'_>,
665        operation_id: OperationId,
666        op_type: &str,
667        operation_meta: impl serde::Serialize + Debug,
668        sms: Vec<DynState>,
669    ) -> anyhow::Result<()> {
670        self.manual_operation_start_inner(
671            &mut dbtx.global_dbtx(self.global_dbtx_access_token),
672            operation_id,
673            op_type,
674            operation_meta,
675            sms,
676        )
677        .await
678    }
679
680    /// See [`Self::manual_operation_start`], just inside a database
681    /// transaction.
682    async fn manual_operation_start_inner(
683        &self,
684        dbtx: &mut DatabaseTransaction<'_>,
685        operation_id: OperationId,
686        op_type: &str,
687        operation_meta: impl serde::Serialize + Debug,
688        sms: Vec<DynState>,
689    ) -> anyhow::Result<()> {
690        dbtx.ensure_global()
691            .expect("Must deal with global dbtx here");
692
693        if self
694            .client
695            .get()
696            .operation_log()
697            .get_operation_dbtx(&mut dbtx.to_ref_nc(), operation_id)
698            .await
699            .is_some()
700        {
701            bail!(
702                "Operation with id {} already exists",
703                operation_id.fmt_short()
704            );
705        }
706
707        self.client
708            .get()
709            .operation_log()
710            .add_operation_log_entry_dbtx(
711                &mut dbtx.to_ref_nc(),
712                operation_id,
713                op_type,
714                serde_json::to_value(operation_meta).expect("Can't fail"),
715            )
716            .await;
717
718        self.client
719            .get()
720            .executor()
721            .add_state_machines_dbtx(&mut dbtx.to_ref_nc(), sms)
722            .await
723            .expect("State machine is valid");
724
725        Ok(())
726    }
727
728    /// Resolve an operation to its cached outcome, or to a caching update
729    /// stream built by `stream_gen`.
730    ///
731    /// `is_terminal` reports whether an update is a final state of the
732    /// operation; only terminal updates are cached as the durable outcome (a
733    /// stream that ends early on a non-terminal update caches nothing). The
734    /// same predicate also validates outcomes READ from the cache: a cached
735    /// outcome that fails to deserialize into `U` (e.g. one written by an
736    /// incompatible earlier version) or that is non-terminal (one frozen by a
737    /// previous version that cached whatever update a stream ended on) is
738    /// discarded with a warning and the state is rebuilt from the update
739    /// stream — whose terminal end then overwrites the stale cache — instead
740    /// of short-circuiting subscribers to it or panicking.
741    pub fn outcome_or_updates<U, S>(
742        &self,
743        operation: &OperationLogEntry,
744        operation_id: OperationId,
745        is_terminal: impl Fn(&U) -> bool + MaybeSend + MaybeSync + 'static,
746        stream_gen: impl FnOnce() -> S + MaybeSend + 'static,
747    ) -> UpdateStreamOrOutcome<U>
748    where
749        U: Clone + Serialize + DeserializeOwned + Debug + MaybeSend + MaybeSync + 'static,
750        S: Stream<Item = U> + MaybeSend + 'static,
751    {
752        use futures::StreamExt;
753        match operation.try_outcome::<U>() {
754            Ok(Some(outcome)) if is_terminal(&outcome) => {
755                return UpdateStreamOrOutcome::Outcome(outcome);
756            }
757            Ok(Some(_non_terminal)) => {
758                warn!(
759                    target: LOG_CLIENT,
760                    "Cached operation outcome is not a terminal update (cached by a previous \
761                     version); rebuilding it from the update stream"
762                );
763            }
764            Ok(None) => {}
765            Err(err) => {
766                warn!(
767                    target: LOG_CLIENT,
768                    err = %err.fmt_compact(),
769                    "Cached operation outcome failed to deserialize; rebuilding it from the update stream"
770                );
771            }
772        }
773        let stream = self
774            .client
775            .get()
776            .operation_log()
777            .caching_operation_update_stream(
778                operation_id,
779                Box::new(move || {
780                    let stream_gen = stream_gen();
781                    Box::pin(
782                        stream_gen.map(move |item| serde_json::to_value(item).expect("Can't fail")),
783                    )
784                }),
785                Box::new(move |update| {
786                    // The update was serialized from a `U` moments ago in this
787                    // process, so deserialization only fails if `U` is not
788                    // round-trip-safe; treat that conservatively as non-terminal
789                    // (skip caching) rather than panicking inside the stream.
790                    serde_json::from_value::<U>(update.clone())
791                        .map(|update| is_terminal(&update))
792                        .unwrap_or(false)
793                }),
794            );
795        UpdateStreamOrOutcome::UpdateStream(Box::pin(
796            stream.map(|u| serde_json::from_value::<U>(u).expect("Can't fail")),
797        ))
798    }
799
800    pub async fn claim_inputs<I, S>(
801        &self,
802        dbtx: &mut DatabaseTransaction<'_>,
803        inputs: ClientInputBundle<I, S>,
804        operation_id: OperationId,
805    ) -> anyhow::Result<OutPointRange>
806    where
807        I: IInput + MaybeSend + MaybeSync + 'static,
808        S: sm::IState + MaybeSend + MaybeSync + 'static,
809    {
810        self.claim_inputs_dyn(dbtx, inputs.into_instanceless(), operation_id)
811            .await
812    }
813
814    async fn claim_inputs_dyn(
815        &self,
816        dbtx: &mut DatabaseTransaction<'_>,
817        inputs: InstancelessDynClientInputBundle,
818        operation_id: OperationId,
819    ) -> anyhow::Result<OutPointRange> {
820        let tx_builder =
821            TransactionBuilder::new().with_inputs(inputs.into_dyn(self.module_instance_id));
822
823        self.client
824            .get()
825            .finalize_and_submit_transaction_inner(
826                &mut dbtx.global_dbtx(self.global_dbtx_access_token),
827                operation_id,
828                tx_builder,
829            )
830            .await
831    }
832
833    pub async fn add_state_machines_dbtx(
834        &self,
835        dbtx: &mut DatabaseTransaction<'_>,
836        states: Vec<DynState>,
837    ) -> AddStateMachinesResult {
838        self.client
839            .get()
840            .executor()
841            .add_state_machines_dbtx(&mut dbtx.global_dbtx(self.global_dbtx_access_token), states)
842            .await
843    }
844
845    /// Reads an operation log entry within an ongoing database transaction.
846    ///
847    /// Unlike [`Self::get_operation`] this observes writes made earlier in
848    /// `dbtx`, which is what a caller needs to decide atomically whether it is
849    /// about to create an operation that already exists.
850    pub async fn get_operation_dbtx(
851        &self,
852        dbtx: &mut DatabaseTransaction<'_>,
853        operation_id: OperationId,
854    ) -> Option<oplog::OperationLogEntry> {
855        self.client
856            .get()
857            .operation_log()
858            .get_operation_dbtx(
859                &mut dbtx.global_dbtx(self.global_dbtx_access_token),
860                operation_id,
861            )
862            .await
863    }
864
865    pub async fn add_operation_log_entry_dbtx(
866        &self,
867        dbtx: &mut DatabaseTransaction<'_>,
868        operation_id: OperationId,
869        operation_type: &str,
870        operation_meta: impl serde::Serialize,
871    ) {
872        self.client
873            .get()
874            .operation_log()
875            .add_operation_log_entry_dbtx(
876                &mut dbtx.global_dbtx(self.global_dbtx_access_token),
877                operation_id,
878                operation_type,
879                serde_json::to_value(operation_meta).expect("Can't fail"),
880            )
881            .await;
882    }
883
884    pub async fn log_event<E, Cap>(&self, dbtx: &mut DatabaseTransaction<'_, Cap>, event: E)
885    where
886        E: Event + Send,
887        Cap: Send,
888    {
889        if <E as Event>::MODULE != Some(<M as ClientModule>::kind()) {
890            warn!(
891                target: LOG_CLIENT,
892                module_kind = %<M as ClientModule>::kind(),
893                event_module = ?<E as Event>::MODULE,
894                "Client module logging events of different module than its own. This might become an error in the future."
895            );
896        }
897        self.client
898            .get()
899            .log_event_json(
900                &mut dbtx.global_dbtx(self.global_dbtx_access_token).to_ref_nc(),
901                <E as Event>::MODULE,
902                self.module_instance_id,
903                <E as Event>::KIND,
904                serde_json::to_value(event).expect("Can't fail"),
905                <E as Event>::PERSISTENCE,
906            )
907            .await;
908    }
909}
910
911/// Priority module priority (lower number is higher priority)
912#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
913pub struct PrimaryModulePriority(u64);
914
915impl PrimaryModulePriority {
916    pub const HIGH: Self = Self(100);
917    pub const LOW: Self = Self(10000);
918
919    pub fn custom(prio: u64) -> Self {
920        Self(prio)
921    }
922}
923/// Which amount units this module supports being primary for
924pub enum PrimaryModuleSupport {
925    /// Potentially any unit
926    Any { priority: PrimaryModulePriority },
927    /// Some units supported by the module
928    Selected {
929        priority: PrimaryModulePriority,
930        units: BTreeSet<AmountUnit>,
931    },
932    /// None
933    None,
934}
935
936impl PrimaryModuleSupport {
937    pub fn selected<const N: usize>(
938        priority: PrimaryModulePriority,
939        units: [AmountUnit; N],
940    ) -> Self {
941        Self::Selected {
942            priority,
943            units: BTreeSet::from(units),
944        }
945    }
946}
947
948/// Fedimint module client
949#[apply(async_trait_maybe_send!)]
950pub trait ClientModule: Debug + MaybeSend + MaybeSync + 'static {
951    type Init: ClientModuleInit;
952
953    /// Common module types shared between client and server
954    type Common: ModuleCommon;
955
956    /// Data stored in regular backups so that restoring doesn't have to start
957    /// from epoch 0
958    type Backup: ModuleBackup;
959
960    /// Data and API clients available to state machine transitions of this
961    /// module
962    type ModuleStateMachineContext: Context;
963
964    /// All possible states this client can submit to the executor
965    type States: State<ModuleContext = Self::ModuleStateMachineContext>
966        + IntoDynInstance<DynType = DynState>;
967
968    fn decoder() -> Decoder {
969        let mut decoder_builder = Self::Common::decoder_builder();
970        decoder_builder.with_decodable_type::<Self::States>();
971        decoder_builder.with_decodable_type::<Self::Backup>();
972        decoder_builder.build()
973    }
974
975    fn kind() -> ModuleKind {
976        <<<Self as ClientModule>::Init as ModuleInit>::Common as CommonModuleInit>::KIND
977    }
978
979    fn context(&self) -> Self::ModuleStateMachineContext;
980
981    /// Initialize client.
982    ///
983    /// Called by the core client code on start, after [`ClientContext`] is
984    /// fully initialized, so unlike during [`ClientModuleInit::init`],
985    /// access to global client is allowed.
986    async fn start(&self) {}
987
988    async fn handle_cli_command(
989        &self,
990        _args: &[ffi::OsString],
991    ) -> anyhow::Result<serde_json::Value> {
992        Err(anyhow::format_err!(
993            "This module does not implement cli commands"
994        ))
995    }
996
997    async fn handle_rpc(
998        &self,
999        _method: String,
1000        _request: serde_json::Value,
1001    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>> {
1002        Box::pin(futures::stream::once(std::future::ready(Err(
1003            anyhow::format_err!("This module does not implement rpc"),
1004        ))))
1005    }
1006
1007    /// Returns the fee the processing of this input requires.
1008    ///
1009    /// If the semantics of a given input aren't known this function returns
1010    /// `None`, this only happens if a future version of Fedimint introduces a
1011    /// new input variant. For clients this should only be the case when
1012    /// processing transactions created by other users, so the result of
1013    /// this function can be `unwrap`ped whenever dealing with inputs
1014    /// generated by ourselves.
1015    fn input_fee(
1016        &self,
1017        amount: &Amounts,
1018        input: &<Self::Common as ModuleCommon>::Input,
1019    ) -> Option<Amounts>;
1020
1021    /// Returns the fee the processing of this output requires.
1022    ///
1023    /// If the semantics of a given output aren't known this function returns
1024    /// `None`, this only happens if a future version of Fedimint introduces a
1025    /// new output variant. For clients this should only be the case when
1026    /// processing transactions created by other users, so the result of
1027    /// this function can be `unwrap`ped whenever dealing with inputs
1028    /// generated by ourselves.
1029    fn output_fee(
1030        &self,
1031        amount: &Amounts,
1032        output: &<Self::Common as ModuleCommon>::Output,
1033    ) -> Option<Amounts>;
1034
1035    fn supports_backup(&self) -> bool {
1036        false
1037    }
1038
1039    async fn backup(&self) -> anyhow::Result<Self::Backup> {
1040        anyhow::bail!("Backup not supported");
1041    }
1042
1043    /// Does this module support being a primary module
1044    ///
1045    /// If it does it must implement:
1046    ///
1047    /// * [`Self::create_final_inputs_and_outputs`]
1048    /// * [`Self::await_primary_module_output`]
1049    /// * [`Self::get_balance`]
1050    /// * [`Self::subscribe_balance_changes`]
1051    fn supports_being_primary(&self) -> PrimaryModuleSupport {
1052        PrimaryModuleSupport::None
1053    }
1054
1055    /// Creates all inputs and outputs necessary to balance the transaction.
1056    /// The function returns an error if and only if the client's funds are not
1057    /// sufficient to create the inputs necessary to fully fund the transaction.
1058    ///
1059    /// A returned input also contains:
1060    /// * A set of private keys belonging to the input for signing the
1061    ///   transaction
1062    /// * A closure that generates states belonging to the input. This closure
1063    ///   takes the transaction id of the transaction in which the input was
1064    ///   used and the input index as input since these cannot be known at time
1065    ///   of calling `create_funding_input` and have to be injected later.
1066    ///
1067    /// A returned output also contains:
1068    /// * A closure that generates states belonging to the output. This closure
1069    ///   takes the transaction id of the transaction in which the output was
1070    ///   used and the output index as input since these cannot be known at time
1071    ///   of calling `create_change_output` and have to be injected later.
1072    async fn create_final_inputs_and_outputs(
1073        &self,
1074        _dbtx: &mut DatabaseTransaction<'_>,
1075        _operation_id: OperationId,
1076        _unit: AmountUnit,
1077        _input_amount: Amount,
1078        _output_amount: Amount,
1079    ) -> anyhow::Result<(
1080        ClientInputBundle<<Self::Common as ModuleCommon>::Input, Self::States>,
1081        ClientOutputBundle<<Self::Common as ModuleCommon>::Output, Self::States>,
1082    )> {
1083        unimplemented!()
1084    }
1085
1086    /// Waits for the funds from an output created by
1087    /// [`Self::create_final_inputs_and_outputs`] to become available. This
1088    /// function returning typically implies a change in the output of
1089    /// [`Self::get_balance`].
1090    async fn await_primary_module_output(
1091        &self,
1092        _operation_id: OperationId,
1093        _out_point: OutPoint,
1094    ) -> anyhow::Result<()> {
1095        unimplemented!()
1096    }
1097
1098    /// Returns the balance held by this module and available for funding
1099    /// transactions.
1100    async fn get_balance(&self, _dbtx: &mut DatabaseTransaction<'_>, _unit: AmountUnit) -> Amount {
1101        unimplemented!()
1102    }
1103
1104    /// Returns the balance held by this module and available for funding
1105    /// transactions.
1106    async fn get_balances(&self, _dbtx: &mut DatabaseTransaction<'_>) -> Amounts {
1107        unimplemented!()
1108    }
1109
1110    /// Returns a stream that will output the updated module balance each time
1111    /// it changes.
1112    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()> {
1113        unimplemented!()
1114    }
1115
1116    /// Leave the federation
1117    ///
1118    /// While technically there's nothing stopping the client from just
1119    /// abandoning Federation at any point by deleting all the related
1120    /// local data, it is useful to make sure it's safe beforehand.
1121    ///
1122    /// This call indicates the desire of the caller client code
1123    /// to orderly and safely leave the Federation by this module instance.
1124    /// The goal of the implementations is to fulfil that wish,
1125    /// giving prompt and informative feedback if it's not yet possible.
1126    ///
1127    /// The client module implementation should handle the request
1128    /// and return as fast as possible avoiding blocking for longer than
1129    /// necessary. This would usually involve some combination of:
1130    ///
1131    /// * recording the state of being in process of leaving the Federation to
1132    ///   prevent initiating new conditions that could delay its completion;
1133    /// * performing any fast to complete cleanup/exit logic;
1134    /// * initiating any time-consuming logic (e.g. canceling outstanding
1135    ///   contracts), as background jobs, tasks machines, etc.
1136    /// * checking for any conditions indicating it might not be safe to leave
1137    ///   at the moment.
1138    ///
1139    /// This function should return `Ok` only if from the perspective
1140    /// of this module instance, it is safe to delete client data and
1141    /// stop using it, with no further actions (like background jobs) required
1142    /// to complete.
1143    ///
1144    /// This function should return an error if it's not currently possible
1145    /// to safely (e.g. without losing funds) leave the Federation.
1146    /// It should avoid running indefinitely trying to complete any cleanup
1147    /// actions necessary to reach a clean state, preferring spawning new
1148    /// state machines and returning an informative error about cleanup
1149    /// still in progress.
1150    ///
1151    /// If any internal task needs to complete, any user action is required,
1152    /// or even external condition needs to be met this function
1153    /// should return a `Err`.
1154    ///
1155    /// Notably modules should not disable interaction that might be necessary
1156    /// for the user (possibly through other modules) to leave the Federation.
1157    /// In particular a Mint module should retain ability to create new notes,
1158    /// and LN module should retain ability to send funds out.
1159    ///
1160    /// Calling code must NOT assume that a module that once returned `Ok`,
1161    /// will not return `Err` at later point. E.g. a Mint module might have
1162    /// no outstanding balance at first, but other modules winding down
1163    /// might "cash-out" to Ecash.
1164    ///
1165    /// Before leaving the Federation and deleting any state the calling code
1166    /// must collect a full round of `Ok` from all the modules.
1167    ///
1168    /// Calling code should allow the user to override and ignore any
1169    /// outstanding errors, after sufficient amount of warnings. Ideally,
1170    /// this should be done on per-module basis, to avoid mistakes.
1171    async fn leave(&self, _dbtx: &mut DatabaseTransaction<'_>) -> anyhow::Result<()> {
1172        bail!("Unable to determine if safe to leave the federation: Not implemented")
1173    }
1174}
1175
1176/// Type-erased version of [`ClientModule`]
1177#[apply(async_trait_maybe_send!)]
1178pub trait IClientModule: Debug {
1179    fn as_any(&self) -> &(maybe_add_send_sync!(dyn std::any::Any));
1180
1181    fn as_any_arc(self: Arc<Self>) -> Arc<maybe_add_send_sync!(dyn std::any::Any + 'static)>;
1182
1183    fn decoder(&self) -> Decoder;
1184
1185    fn context(&self, instance: ModuleInstanceId) -> DynContext;
1186
1187    async fn start(&self);
1188
1189    async fn handle_cli_command(&self, args: &[ffi::OsString])
1190    -> anyhow::Result<serde_json::Value>;
1191
1192    async fn handle_rpc(
1193        &self,
1194        method: String,
1195        request: serde_json::Value,
1196    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>>;
1197
1198    fn input_fee(&self, amount: &Amounts, input: &DynInput) -> Option<Amounts>;
1199
1200    fn output_fee(&self, amount: &Amounts, output: &DynOutput) -> Option<Amounts>;
1201
1202    fn supports_backup(&self) -> bool;
1203
1204    async fn backup(&self, module_instance_id: ModuleInstanceId)
1205    -> anyhow::Result<DynModuleBackup>;
1206
1207    fn supports_being_primary(&self) -> PrimaryModuleSupport;
1208
1209    async fn create_final_inputs_and_outputs(
1210        &self,
1211        module_instance: ModuleInstanceId,
1212        dbtx: &mut DatabaseTransaction<'_>,
1213        operation_id: OperationId,
1214        unit: AmountUnit,
1215        input_amount: Amount,
1216        output_amount: Amount,
1217    ) -> anyhow::Result<(ClientInputBundle, ClientOutputBundle)>;
1218
1219    async fn await_primary_module_output(
1220        &self,
1221        operation_id: OperationId,
1222        out_point: OutPoint,
1223    ) -> anyhow::Result<()>;
1224
1225    async fn get_balance(
1226        &self,
1227        module_instance: ModuleInstanceId,
1228        dbtx: &mut DatabaseTransaction<'_>,
1229        unit: AmountUnit,
1230    ) -> Amount;
1231
1232    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()>;
1233}
1234
1235#[apply(async_trait_maybe_send!)]
1236impl<T> IClientModule for T
1237where
1238    T: ClientModule,
1239{
1240    fn as_any(&self) -> &(maybe_add_send_sync!(dyn Any)) {
1241        self
1242    }
1243
1244    fn as_any_arc(self: Arc<Self>) -> Arc<maybe_add_send_sync!(dyn Any + 'static)> {
1245        self
1246    }
1247
1248    fn decoder(&self) -> Decoder {
1249        T::decoder()
1250    }
1251
1252    fn context(&self, instance: ModuleInstanceId) -> DynContext {
1253        DynContext::from_typed(instance, <T as ClientModule>::context(self))
1254    }
1255
1256    async fn start(&self) {
1257        <T as ClientModule>::start(self).await;
1258    }
1259
1260    async fn handle_cli_command(
1261        &self,
1262        args: &[ffi::OsString],
1263    ) -> anyhow::Result<serde_json::Value> {
1264        <T as ClientModule>::handle_cli_command(self, args).await
1265    }
1266
1267    async fn handle_rpc(
1268        &self,
1269        method: String,
1270        request: serde_json::Value,
1271    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>> {
1272        <T as ClientModule>::handle_rpc(self, method, request).await
1273    }
1274
1275    fn input_fee(&self, amount: &Amounts, input: &DynInput) -> Option<Amounts> {
1276        <T as ClientModule>::input_fee(
1277            self,
1278            amount,
1279            input
1280                .as_any()
1281                .downcast_ref()
1282                .expect("Dispatched to correct module"),
1283        )
1284    }
1285
1286    fn output_fee(&self, amount: &Amounts, output: &DynOutput) -> Option<Amounts> {
1287        <T as ClientModule>::output_fee(
1288            self,
1289            amount,
1290            output
1291                .as_any()
1292                .downcast_ref()
1293                .expect("Dispatched to correct module"),
1294        )
1295    }
1296
1297    fn supports_backup(&self) -> bool {
1298        <T as ClientModule>::supports_backup(self)
1299    }
1300
1301    async fn backup(
1302        &self,
1303        module_instance_id: ModuleInstanceId,
1304    ) -> anyhow::Result<DynModuleBackup> {
1305        Ok(DynModuleBackup::from_typed(
1306            module_instance_id,
1307            <T as ClientModule>::backup(self).await?,
1308        ))
1309    }
1310
1311    fn supports_being_primary(&self) -> PrimaryModuleSupport {
1312        <T as ClientModule>::supports_being_primary(self)
1313    }
1314
1315    async fn create_final_inputs_and_outputs(
1316        &self,
1317        module_instance: ModuleInstanceId,
1318        dbtx: &mut DatabaseTransaction<'_>,
1319        operation_id: OperationId,
1320        unit: AmountUnit,
1321        input_amount: Amount,
1322        output_amount: Amount,
1323    ) -> anyhow::Result<(ClientInputBundle, ClientOutputBundle)> {
1324        let (inputs, outputs) = <T as ClientModule>::create_final_inputs_and_outputs(
1325            self,
1326            &mut dbtx.to_ref_with_prefix_module_id(module_instance).0,
1327            operation_id,
1328            unit,
1329            input_amount,
1330            output_amount,
1331        )
1332        .await?;
1333
1334        let inputs = inputs.into_dyn(module_instance);
1335
1336        let outputs = outputs.into_dyn(module_instance);
1337
1338        Ok((inputs, outputs))
1339    }
1340
1341    async fn await_primary_module_output(
1342        &self,
1343        operation_id: OperationId,
1344        out_point: OutPoint,
1345    ) -> anyhow::Result<()> {
1346        <T as ClientModule>::await_primary_module_output(self, operation_id, out_point).await
1347    }
1348
1349    async fn get_balance(
1350        &self,
1351        module_instance: ModuleInstanceId,
1352        dbtx: &mut DatabaseTransaction<'_>,
1353        unit: AmountUnit,
1354    ) -> Amount {
1355        <T as ClientModule>::get_balance(
1356            self,
1357            &mut dbtx.to_ref_with_prefix_module_id(module_instance).0,
1358            unit,
1359        )
1360        .await
1361    }
1362
1363    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()> {
1364        <T as ClientModule>::subscribe_balance_changes(self).await
1365    }
1366}
1367
1368dyn_newtype_define!(
1369    #[derive(Clone)]
1370    pub DynClientModule(Arc<IClientModule>)
1371);
1372
1373impl DynClientModule {
1374    pub fn as_any_arc(&self) -> Arc<maybe_add_send_sync!(dyn Any + 'static)> {
1375        self.inner.clone().as_any_arc()
1376    }
1377}
1378
1379impl AsRef<maybe_add_send_sync!(dyn IClientModule + 'static)> for DynClientModule {
1380    fn as_ref(&self) -> &maybe_add_send_sync!(dyn IClientModule + 'static) {
1381        self.inner.as_ref()
1382    }
1383}
1384
1385// Re-export types from fedimint_core
1386pub use fedimint_core::{IdxRange, OutPointRange, OutPointRangeIter};
1387
1388pub type StateGenerator<S> = Arc<maybe_add_send_sync!(dyn Fn(OutPointRange) -> Vec<S> + 'static)>;
fedimint_client_module/module/mod.rs

fedimint_client_module/module/
mod.rs
