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;
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    pub fn outcome_or_updates<U, S>(
729        &self,
730        operation: OperationLogEntry,
731        operation_id: OperationId,
732        stream_gen: impl FnOnce() -> S + 'static,
733    ) -> UpdateStreamOrOutcome<U>
734    where
735        U: Clone + Serialize + DeserializeOwned + Debug + MaybeSend + MaybeSync + 'static,
736        S: Stream<Item = U> + MaybeSend + 'static,
737    {
738        use futures::StreamExt;
739        match self.client.get().operation_log().outcome_or_updates(
740            &self.global_db(),
741            operation_id,
742            operation,
743            Box::new(move || {
744                let stream_gen = stream_gen();
745                Box::pin(
746                    stream_gen.map(move |item| serde_json::to_value(item).expect("Can't fail")),
747                )
748            }),
749        ) {
750            UpdateStreamOrOutcome::UpdateStream(stream) => UpdateStreamOrOutcome::UpdateStream(
751                Box::pin(stream.map(|u| serde_json::from_value(u).expect("Can't fail"))),
752            ),
753            UpdateStreamOrOutcome::Outcome(o) => {
754                UpdateStreamOrOutcome::Outcome(serde_json::from_value(o).expect("Can't fail"))
755            }
756        }
757    }
758
759    pub async fn claim_inputs<I, S>(
760        &self,
761        dbtx: &mut DatabaseTransaction<'_>,
762        inputs: ClientInputBundle<I, S>,
763        operation_id: OperationId,
764    ) -> anyhow::Result<OutPointRange>
765    where
766        I: IInput + MaybeSend + MaybeSync + 'static,
767        S: sm::IState + MaybeSend + MaybeSync + 'static,
768    {
769        self.claim_inputs_dyn(dbtx, inputs.into_instanceless(), operation_id)
770            .await
771    }
772
773    async fn claim_inputs_dyn(
774        &self,
775        dbtx: &mut DatabaseTransaction<'_>,
776        inputs: InstancelessDynClientInputBundle,
777        operation_id: OperationId,
778    ) -> anyhow::Result<OutPointRange> {
779        let tx_builder =
780            TransactionBuilder::new().with_inputs(inputs.into_dyn(self.module_instance_id));
781
782        self.client
783            .get()
784            .finalize_and_submit_transaction_inner(
785                &mut dbtx.global_dbtx(self.global_dbtx_access_token),
786                operation_id,
787                tx_builder,
788            )
789            .await
790    }
791
792    pub async fn add_state_machines_dbtx(
793        &self,
794        dbtx: &mut DatabaseTransaction<'_>,
795        states: Vec<DynState>,
796    ) -> AddStateMachinesResult {
797        self.client
798            .get()
799            .executor()
800            .add_state_machines_dbtx(&mut dbtx.global_dbtx(self.global_dbtx_access_token), states)
801            .await
802    }
803
804    pub async fn add_operation_log_entry_dbtx(
805        &self,
806        dbtx: &mut DatabaseTransaction<'_>,
807        operation_id: OperationId,
808        operation_type: &str,
809        operation_meta: impl serde::Serialize,
810    ) {
811        self.client
812            .get()
813            .operation_log()
814            .add_operation_log_entry_dbtx(
815                &mut dbtx.global_dbtx(self.global_dbtx_access_token),
816                operation_id,
817                operation_type,
818                serde_json::to_value(operation_meta).expect("Can't fail"),
819            )
820            .await;
821    }
822
823    pub async fn log_event<E, Cap>(&self, dbtx: &mut DatabaseTransaction<'_, Cap>, event: E)
824    where
825        E: Event + Send,
826        Cap: Send,
827    {
828        if <E as Event>::MODULE != Some(<M as ClientModule>::kind()) {
829            warn!(
830                target: LOG_CLIENT,
831                module_kind = %<M as ClientModule>::kind(),
832                event_module = ?<E as Event>::MODULE,
833                "Client module logging events of different module than its own. This might become an error in the future."
834            );
835        }
836        self.client
837            .get()
838            .log_event_json(
839                &mut dbtx.global_dbtx(self.global_dbtx_access_token).to_ref_nc(),
840                <E as Event>::MODULE,
841                self.module_instance_id,
842                <E as Event>::KIND,
843                serde_json::to_value(event).expect("Can't fail"),
844                <E as Event>::PERSISTENCE,
845            )
846            .await;
847    }
848}
849
850/// Priority module priority (lower number is higher priority)
851#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
852pub struct PrimaryModulePriority(u64);
853
854impl PrimaryModulePriority {
855    pub const HIGH: Self = Self(100);
856    pub const LOW: Self = Self(10000);
857
858    pub fn custom(prio: u64) -> Self {
859        Self(prio)
860    }
861}
862/// Which amount units this module supports being primary for
863pub enum PrimaryModuleSupport {
864    /// Potentially any unit
865    Any { priority: PrimaryModulePriority },
866    /// Some units supported by the module
867    Selected {
868        priority: PrimaryModulePriority,
869        units: BTreeSet<AmountUnit>,
870    },
871    /// None
872    None,
873}
874
875impl PrimaryModuleSupport {
876    pub fn selected<const N: usize>(
877        priority: PrimaryModulePriority,
878        units: [AmountUnit; N],
879    ) -> Self {
880        Self::Selected {
881            priority,
882            units: BTreeSet::from(units),
883        }
884    }
885}
886
887/// Fedimint module client
888#[apply(async_trait_maybe_send!)]
889pub trait ClientModule: Debug + MaybeSend + MaybeSync + 'static {
890    type Init: ClientModuleInit;
891
892    /// Common module types shared between client and server
893    type Common: ModuleCommon;
894
895    /// Data stored in regular backups so that restoring doesn't have to start
896    /// from epoch 0
897    type Backup: ModuleBackup;
898
899    /// Data and API clients available to state machine transitions of this
900    /// module
901    type ModuleStateMachineContext: Context;
902
903    /// All possible states this client can submit to the executor
904    type States: State<ModuleContext = Self::ModuleStateMachineContext>
905        + IntoDynInstance<DynType = DynState>;
906
907    fn decoder() -> Decoder {
908        let mut decoder_builder = Self::Common::decoder_builder();
909        decoder_builder.with_decodable_type::<Self::States>();
910        decoder_builder.with_decodable_type::<Self::Backup>();
911        decoder_builder.build()
912    }
913
914    fn kind() -> ModuleKind {
915        <<<Self as ClientModule>::Init as ModuleInit>::Common as CommonModuleInit>::KIND
916    }
917
918    fn context(&self) -> Self::ModuleStateMachineContext;
919
920    /// Initialize client.
921    ///
922    /// Called by the core client code on start, after [`ClientContext`] is
923    /// fully initialized, so unlike during [`ClientModuleInit::init`],
924    /// access to global client is allowed.
925    async fn start(&self) {}
926
927    async fn handle_cli_command(
928        &self,
929        _args: &[ffi::OsString],
930    ) -> anyhow::Result<serde_json::Value> {
931        Err(anyhow::format_err!(
932            "This module does not implement cli commands"
933        ))
934    }
935
936    async fn handle_rpc(
937        &self,
938        _method: String,
939        _request: serde_json::Value,
940    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>> {
941        Box::pin(futures::stream::once(std::future::ready(Err(
942            anyhow::format_err!("This module does not implement rpc"),
943        ))))
944    }
945
946    /// Returns the fee the processing of this input requires.
947    ///
948    /// If the semantics of a given input aren't known this function returns
949    /// `None`, this only happens if a future version of Fedimint introduces a
950    /// new input variant. For clients this should only be the case when
951    /// processing transactions created by other users, so the result of
952    /// this function can be `unwrap`ped whenever dealing with inputs
953    /// generated by ourselves.
954    fn input_fee(
955        &self,
956        amount: &Amounts,
957        input: &<Self::Common as ModuleCommon>::Input,
958    ) -> Option<Amounts>;
959
960    /// Returns the fee the processing of this output requires.
961    ///
962    /// If the semantics of a given output aren't known this function returns
963    /// `None`, this only happens if a future version of Fedimint introduces a
964    /// new output variant. For clients this should only be the case when
965    /// processing transactions created by other users, so the result of
966    /// this function can be `unwrap`ped whenever dealing with inputs
967    /// generated by ourselves.
968    fn output_fee(
969        &self,
970        amount: &Amounts,
971        output: &<Self::Common as ModuleCommon>::Output,
972    ) -> Option<Amounts>;
973
974    fn supports_backup(&self) -> bool {
975        false
976    }
977
978    async fn backup(&self) -> anyhow::Result<Self::Backup> {
979        anyhow::bail!("Backup not supported");
980    }
981
982    /// Does this module support being a primary module
983    ///
984    /// If it does it must implement:
985    ///
986    /// * [`Self::create_final_inputs_and_outputs`]
987    /// * [`Self::await_primary_module_output`]
988    /// * [`Self::get_balance`]
989    /// * [`Self::subscribe_balance_changes`]
990    fn supports_being_primary(&self) -> PrimaryModuleSupport {
991        PrimaryModuleSupport::None
992    }
993
994    /// Creates all inputs and outputs necessary to balance the transaction.
995    /// The function returns an error if and only if the client's funds are not
996    /// sufficient to create the inputs necessary to fully fund the transaction.
997    ///
998    /// A returned input also contains:
999    /// * A set of private keys belonging to the input for signing the
1000    ///   transaction
1001    /// * A closure that generates states belonging to the input. This closure
1002    ///   takes the transaction id of the transaction in which the input was
1003    ///   used and the input index as input since these cannot be known at time
1004    ///   of calling `create_funding_input` and have to be injected later.
1005    ///
1006    /// A returned output also contains:
1007    /// * A closure that generates states belonging to the output. This closure
1008    ///   takes the transaction id of the transaction in which the output was
1009    ///   used and the output index as input since these cannot be known at time
1010    ///   of calling `create_change_output` and have to be injected later.
1011    async fn create_final_inputs_and_outputs(
1012        &self,
1013        _dbtx: &mut DatabaseTransaction<'_>,
1014        _operation_id: OperationId,
1015        _unit: AmountUnit,
1016        _input_amount: Amount,
1017        _output_amount: Amount,
1018    ) -> anyhow::Result<(
1019        ClientInputBundle<<Self::Common as ModuleCommon>::Input, Self::States>,
1020        ClientOutputBundle<<Self::Common as ModuleCommon>::Output, Self::States>,
1021    )> {
1022        unimplemented!()
1023    }
1024
1025    /// Waits for the funds from an output created by
1026    /// [`Self::create_final_inputs_and_outputs`] to become available. This
1027    /// function returning typically implies a change in the output of
1028    /// [`Self::get_balance`].
1029    async fn await_primary_module_output(
1030        &self,
1031        _operation_id: OperationId,
1032        _out_point: OutPoint,
1033    ) -> anyhow::Result<()> {
1034        unimplemented!()
1035    }
1036
1037    /// Returns the balance held by this module and available for funding
1038    /// transactions.
1039    async fn get_balance(&self, _dbtx: &mut DatabaseTransaction<'_>, _unit: AmountUnit) -> Amount {
1040        unimplemented!()
1041    }
1042
1043    /// Returns the balance held by this module and available for funding
1044    /// transactions.
1045    async fn get_balances(&self, _dbtx: &mut DatabaseTransaction<'_>) -> Amounts {
1046        unimplemented!()
1047    }
1048
1049    /// Returns a stream that will output the updated module balance each time
1050    /// it changes.
1051    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()> {
1052        unimplemented!()
1053    }
1054
1055    /// Leave the federation
1056    ///
1057    /// While technically there's nothing stopping the client from just
1058    /// abandoning Federation at any point by deleting all the related
1059    /// local data, it is useful to make sure it's safe beforehand.
1060    ///
1061    /// This call indicates the desire of the caller client code
1062    /// to orderly and safely leave the Federation by this module instance.
1063    /// The goal of the implementations is to fulfil that wish,
1064    /// giving prompt and informative feedback if it's not yet possible.
1065    ///
1066    /// The client module implementation should handle the request
1067    /// and return as fast as possible avoiding blocking for longer than
1068    /// necessary. This would usually involve some combination of:
1069    ///
1070    /// * recording the state of being in process of leaving the Federation to
1071    ///   prevent initiating new conditions that could delay its completion;
1072    /// * performing any fast to complete cleanup/exit logic;
1073    /// * initiating any time-consuming logic (e.g. canceling outstanding
1074    ///   contracts), as background jobs, tasks machines, etc.
1075    /// * checking for any conditions indicating it might not be safe to leave
1076    ///   at the moment.
1077    ///
1078    /// This function should return `Ok` only if from the perspective
1079    /// of this module instance, it is safe to delete client data and
1080    /// stop using it, with no further actions (like background jobs) required
1081    /// to complete.
1082    ///
1083    /// This function should return an error if it's not currently possible
1084    /// to safely (e.g. without losing funds) leave the Federation.
1085    /// It should avoid running indefinitely trying to complete any cleanup
1086    /// actions necessary to reach a clean state, preferring spawning new
1087    /// state machines and returning an informative error about cleanup
1088    /// still in progress.
1089    ///
1090    /// If any internal task needs to complete, any user action is required,
1091    /// or even external condition needs to be met this function
1092    /// should return a `Err`.
1093    ///
1094    /// Notably modules should not disable interaction that might be necessary
1095    /// for the user (possibly through other modules) to leave the Federation.
1096    /// In particular a Mint module should retain ability to create new notes,
1097    /// and LN module should retain ability to send funds out.
1098    ///
1099    /// Calling code must NOT assume that a module that once returned `Ok`,
1100    /// will not return `Err` at later point. E.g. a Mint module might have
1101    /// no outstanding balance at first, but other modules winding down
1102    /// might "cash-out" to Ecash.
1103    ///
1104    /// Before leaving the Federation and deleting any state the calling code
1105    /// must collect a full round of `Ok` from all the modules.
1106    ///
1107    /// Calling code should allow the user to override and ignore any
1108    /// outstanding errors, after sufficient amount of warnings. Ideally,
1109    /// this should be done on per-module basis, to avoid mistakes.
1110    async fn leave(&self, _dbtx: &mut DatabaseTransaction<'_>) -> anyhow::Result<()> {
1111        bail!("Unable to determine if safe to leave the federation: Not implemented")
1112    }
1113}
1114
1115/// Type-erased version of [`ClientModule`]
1116#[apply(async_trait_maybe_send!)]
1117pub trait IClientModule: Debug {
1118    fn as_any(&self) -> &(maybe_add_send_sync!(dyn std::any::Any));
1119
1120    fn as_any_arc(self: Arc<Self>) -> Arc<maybe_add_send_sync!(dyn std::any::Any + 'static)>;
1121
1122    fn decoder(&self) -> Decoder;
1123
1124    fn context(&self, instance: ModuleInstanceId) -> DynContext;
1125
1126    async fn start(&self);
1127
1128    async fn handle_cli_command(&self, args: &[ffi::OsString])
1129    -> anyhow::Result<serde_json::Value>;
1130
1131    async fn handle_rpc(
1132        &self,
1133        method: String,
1134        request: serde_json::Value,
1135    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>>;
1136
1137    fn input_fee(&self, amount: &Amounts, input: &DynInput) -> Option<Amounts>;
1138
1139    fn output_fee(&self, amount: &Amounts, output: &DynOutput) -> Option<Amounts>;
1140
1141    fn supports_backup(&self) -> bool;
1142
1143    async fn backup(&self, module_instance_id: ModuleInstanceId)
1144    -> anyhow::Result<DynModuleBackup>;
1145
1146    fn supports_being_primary(&self) -> PrimaryModuleSupport;
1147
1148    async fn create_final_inputs_and_outputs(
1149        &self,
1150        module_instance: ModuleInstanceId,
1151        dbtx: &mut DatabaseTransaction<'_>,
1152        operation_id: OperationId,
1153        unit: AmountUnit,
1154        input_amount: Amount,
1155        output_amount: Amount,
1156    ) -> anyhow::Result<(ClientInputBundle, ClientOutputBundle)>;
1157
1158    async fn await_primary_module_output(
1159        &self,
1160        operation_id: OperationId,
1161        out_point: OutPoint,
1162    ) -> anyhow::Result<()>;
1163
1164    async fn get_balance(
1165        &self,
1166        module_instance: ModuleInstanceId,
1167        dbtx: &mut DatabaseTransaction<'_>,
1168        unit: AmountUnit,
1169    ) -> Amount;
1170
1171    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()>;
1172}
1173
1174#[apply(async_trait_maybe_send!)]
1175impl<T> IClientModule for T
1176where
1177    T: ClientModule,
1178{
1179    fn as_any(&self) -> &(maybe_add_send_sync!(dyn Any)) {
1180        self
1181    }
1182
1183    fn as_any_arc(self: Arc<Self>) -> Arc<maybe_add_send_sync!(dyn Any + 'static)> {
1184        self
1185    }
1186
1187    fn decoder(&self) -> Decoder {
1188        T::decoder()
1189    }
1190
1191    fn context(&self, instance: ModuleInstanceId) -> DynContext {
1192        DynContext::from_typed(instance, <T as ClientModule>::context(self))
1193    }
1194
1195    async fn start(&self) {
1196        <T as ClientModule>::start(self).await;
1197    }
1198
1199    async fn handle_cli_command(
1200        &self,
1201        args: &[ffi::OsString],
1202    ) -> anyhow::Result<serde_json::Value> {
1203        <T as ClientModule>::handle_cli_command(self, args).await
1204    }
1205
1206    async fn handle_rpc(
1207        &self,
1208        method: String,
1209        request: serde_json::Value,
1210    ) -> BoxStream<'_, anyhow::Result<serde_json::Value>> {
1211        <T as ClientModule>::handle_rpc(self, method, request).await
1212    }
1213
1214    fn input_fee(&self, amount: &Amounts, input: &DynInput) -> Option<Amounts> {
1215        <T as ClientModule>::input_fee(
1216            self,
1217            amount,
1218            input
1219                .as_any()
1220                .downcast_ref()
1221                .expect("Dispatched to correct module"),
1222        )
1223    }
1224
1225    fn output_fee(&self, amount: &Amounts, output: &DynOutput) -> Option<Amounts> {
1226        <T as ClientModule>::output_fee(
1227            self,
1228            amount,
1229            output
1230                .as_any()
1231                .downcast_ref()
1232                .expect("Dispatched to correct module"),
1233        )
1234    }
1235
1236    fn supports_backup(&self) -> bool {
1237        <T as ClientModule>::supports_backup(self)
1238    }
1239
1240    async fn backup(
1241        &self,
1242        module_instance_id: ModuleInstanceId,
1243    ) -> anyhow::Result<DynModuleBackup> {
1244        Ok(DynModuleBackup::from_typed(
1245            module_instance_id,
1246            <T as ClientModule>::backup(self).await?,
1247        ))
1248    }
1249
1250    fn supports_being_primary(&self) -> PrimaryModuleSupport {
1251        <T as ClientModule>::supports_being_primary(self)
1252    }
1253
1254    async fn create_final_inputs_and_outputs(
1255        &self,
1256        module_instance: ModuleInstanceId,
1257        dbtx: &mut DatabaseTransaction<'_>,
1258        operation_id: OperationId,
1259        unit: AmountUnit,
1260        input_amount: Amount,
1261        output_amount: Amount,
1262    ) -> anyhow::Result<(ClientInputBundle, ClientOutputBundle)> {
1263        let (inputs, outputs) = <T as ClientModule>::create_final_inputs_and_outputs(
1264            self,
1265            &mut dbtx.to_ref_with_prefix_module_id(module_instance).0,
1266            operation_id,
1267            unit,
1268            input_amount,
1269            output_amount,
1270        )
1271        .await?;
1272
1273        let inputs = inputs.into_dyn(module_instance);
1274
1275        let outputs = outputs.into_dyn(module_instance);
1276
1277        Ok((inputs, outputs))
1278    }
1279
1280    async fn await_primary_module_output(
1281        &self,
1282        operation_id: OperationId,
1283        out_point: OutPoint,
1284    ) -> anyhow::Result<()> {
1285        <T as ClientModule>::await_primary_module_output(self, operation_id, out_point).await
1286    }
1287
1288    async fn get_balance(
1289        &self,
1290        module_instance: ModuleInstanceId,
1291        dbtx: &mut DatabaseTransaction<'_>,
1292        unit: AmountUnit,
1293    ) -> Amount {
1294        <T as ClientModule>::get_balance(
1295            self,
1296            &mut dbtx.to_ref_with_prefix_module_id(module_instance).0,
1297            unit,
1298        )
1299        .await
1300    }
1301
1302    async fn subscribe_balance_changes(&self) -> BoxStream<'static, ()> {
1303        <T as ClientModule>::subscribe_balance_changes(self).await
1304    }
1305}
1306
1307dyn_newtype_define!(
1308    #[derive(Clone)]
1309    pub DynClientModule(Arc<IClientModule>)
1310);
1311
1312impl DynClientModule {
1313    pub fn as_any_arc(&self) -> Arc<maybe_add_send_sync!(dyn Any + 'static)> {
1314        self.inner.clone().as_any_arc()
1315    }
1316}
1317
1318impl AsRef<maybe_add_send_sync!(dyn IClientModule + 'static)> for DynClientModule {
1319    fn as_ref(&self) -> &maybe_add_send_sync!(dyn IClientModule + 'static) {
1320        self.inner.as_ref()
1321    }
1322}
1323
1324// Re-export types from fedimint_core
1325pub use fedimint_core::{IdxRange, OutPointRange, OutPointRangeIter};
1326
1327pub 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
