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