fedimint_client_rpc/
lib.rs

1use std::collections::HashMap;
2use std::str::FromStr;
3use std::sync::Arc;
4use std::time::Duration;
5
6use anyhow::Context;
7use async_stream::try_stream;
8use fedimint_bip39::{Bip39RootSecretStrategy, Mnemonic};
9use fedimint_client::module::ClientModule;
10use fedimint_client::secret::RootSecretStrategy;
11use fedimint_client::{ClientHandleArc, RootSecret};
12use fedimint_core::config::FederationId;
13use fedimint_core::db::{Database, IDatabaseTransactionOpsCoreTyped};
14use fedimint_core::encoding::{Decodable, Encodable};
15use fedimint_core::impl_db_record;
16use fedimint_core::invite_code::InviteCode;
17use fedimint_core::task::{MaybeSend, MaybeSync};
18use fedimint_core::util::{BoxFuture, BoxStream};
19use fedimint_derive_secret::{ChildId, DerivableSecret};
20use fedimint_ln_client::{LightningClientInit, LightningClientModule};
21use fedimint_meta_client::MetaClientInit;
22use fedimint_mint_client::{MintClientInit, MintClientModule};
23use fedimint_wallet_client::{WalletClientInit, WalletClientModule};
24use futures::StreamExt;
25use futures::future::{AbortHandle, Abortable};
26use lightning_invoice::Bolt11InvoiceDescriptionRef;
27use rand::thread_rng;
28use serde::{Deserialize, Serialize};
29use serde_json::json;
30use tokio::sync::Mutex;
31use tracing::info;
32
33// Key prefixes for the unified database
34#[repr(u8)]
35#[derive(Clone, Copy, Debug)]
36pub enum DbKeyPrefix {
37    ClientDatabase = 0x00,
38    Mnemonic = 0x01,
39}
40
41#[derive(Debug, Clone, Encodable, Decodable, Eq, PartialEq, Hash)]
42pub struct MnemonicKey;
43
44impl_db_record!(
45    key = MnemonicKey,
46    value = Vec<u8>,
47    db_prefix = DbKeyPrefix::Mnemonic,
48);
49
50#[derive(Serialize, Deserialize)]
51#[serde(rename_all = "snake_case")]
52pub struct RpcRequest {
53    pub request_id: u64,
54    #[serde(flatten)]
55    pub kind: RpcRequestKind,
56}
57
58#[derive(Serialize, Deserialize)]
59#[serde(tag = "type", rename_all = "snake_case")]
60pub enum RpcRequestKind {
61    SetMnemonic {
62        words: Vec<String>,
63    },
64    GenerateMnemonic,
65    GetMnemonic,
66    /// Join federation (requires mnemonic to be set first)
67    JoinFederation {
68        invite_code: String,
69        force_recover: bool,
70        client_name: String,
71    },
72    OpenClient {
73        client_name: String,
74    },
75    CloseClient {
76        client_name: String,
77    },
78    ClientRpc {
79        client_name: String,
80        module: String,
81        method: String,
82        payload: serde_json::Value,
83    },
84    CancelRpc {
85        cancel_request_id: u64,
86    },
87    ParseInviteCode {
88        invite_code: String,
89    },
90    ParseBolt11Invoice {
91        invoice: String,
92    },
93    PreviewFederation {
94        invite_code: String,
95    },
96}
97
98#[derive(Serialize, Deserialize, Clone, Debug)]
99pub struct RpcResponse {
100    pub request_id: u64,
101    #[serde(flatten)]
102    pub kind: RpcResponseKind,
103}
104
105#[derive(Serialize, Deserialize, Clone, Debug)]
106#[serde(tag = "type", rename_all = "snake_case")]
107pub enum RpcResponseKind {
108    Data { data: serde_json::Value },
109    Error { error: String },
110    Aborted {},
111    End {},
112}
113
114pub trait RpcResponseHandler: MaybeSend + MaybeSync {
115    fn handle_response(&self, response: RpcResponse);
116}
117
118pub struct RpcGlobalState {
119    clients: Mutex<HashMap<String, ClientHandleArc>>,
120    rpc_handles: std::sync::Mutex<HashMap<u64, AbortHandle>>,
121    unified_database: Database,
122}
123
124pub struct HandledRpc<'a> {
125    pub task: Option<BoxFuture<'a, ()>>,
126}
127
128impl RpcGlobalState {
129    pub fn new(unified_database: Database) -> Self {
130        Self {
131            clients: Mutex::new(HashMap::new()),
132            rpc_handles: std::sync::Mutex::new(HashMap::new()),
133            unified_database,
134        }
135    }
136
137    async fn add_client(&self, client_name: String, client: ClientHandleArc) {
138        let mut clients = self.clients.lock().await;
139        clients.insert(client_name, client);
140    }
141
142    async fn get_client(&self, client_name: &str) -> Option<ClientHandleArc> {
143        let clients = self.clients.lock().await;
144        clients.get(client_name).cloned()
145    }
146
147    fn add_rpc_handle(&self, request_id: u64, handle: AbortHandle) {
148        let mut handles = self.rpc_handles.lock().unwrap();
149        if handles.insert(request_id, handle).is_some() {
150            tracing::error!("RPC CLIENT ERROR: request id reuse detected");
151        }
152    }
153
154    fn remove_rpc_handle(&self, request_id: u64) -> Option<AbortHandle> {
155        let mut handles = self.rpc_handles.lock().unwrap();
156        handles.remove(&request_id)
157    }
158
159    async fn client_builder(db: Database) -> Result<fedimint_client::ClientBuilder, anyhow::Error> {
160        let mut builder = fedimint_client::Client::builder(db).await?;
161        builder.with_module(MintClientInit);
162        builder.with_module(LightningClientInit::default());
163        builder.with_module(WalletClientInit(None));
164        builder.with_module(MetaClientInit);
165        builder.with_primary_module_kind(fedimint_mint_client::KIND);
166        Ok(builder)
167    }
168
169    /// Get client-specific database with proper prefix
170    async fn client_db(&self, client_name: String) -> anyhow::Result<Database> {
171        assert_eq!(client_name.len(), 36);
172
173        let unified_db = &self.unified_database;
174        let mut client_prefix = vec![DbKeyPrefix::ClientDatabase as u8];
175        client_prefix.extend_from_slice(client_name.as_bytes());
176        Ok(unified_db.with_prefix(client_prefix))
177    }
178
179    /// Handle joining federation using unified database
180    async fn handle_join_federation(
181        &self,
182        invite_code: String,
183        client_name: String,
184        force_recover: bool,
185    ) -> anyhow::Result<()> {
186        // Check if wallet mnemonic is set
187        let mnemonic = self
188            .get_mnemonic_from_db()
189            .await?
190            .context("No wallet mnemonic set. Please set or generate a mnemonic first.")?;
191
192        let client_db = self.client_db(client_name.clone()).await?;
193
194        let invite_code = InviteCode::from_str(&invite_code)?;
195        let federation_id = invite_code.federation_id();
196
197        // Derive federation-specific secret from wallet mnemonic
198        let federation_secret = self.derive_federation_secret(&mnemonic, &federation_id);
199
200        let builder = Self::client_builder(client_db).await?;
201        let preview = builder.preview(&invite_code).await?;
202
203        // Check if backup exists
204        let backup = preview
205            .download_backup_from_federation(RootSecret::StandardDoubleDerive(
206                federation_secret.clone(),
207            ))
208            .await?;
209
210        let client = if force_recover || backup.is_some() {
211            Arc::new(
212                preview
213                    .recover(RootSecret::StandardDoubleDerive(federation_secret), backup)
214                    .await?,
215            )
216        } else {
217            Arc::new(
218                preview
219                    .join(RootSecret::StandardDoubleDerive(federation_secret))
220                    .await?,
221            )
222        };
223
224        self.add_client(client_name, client).await;
225        Ok(())
226    }
227
228    async fn handle_open_client(&self, client_name: String) -> anyhow::Result<()> {
229        // Check if wallet mnemonic is set
230        let mnemonic = self
231            .get_mnemonic_from_db()
232            .await?
233            .context("No wallet mnemonic set. Please set or generate a mnemonic first.")?;
234
235        let client_db = self.client_db(client_name.clone()).await?;
236
237        if !fedimint_client::Client::is_initialized(&client_db).await {
238            anyhow::bail!("client is not initialized for this database");
239        }
240
241        // Get the client config to retrieve the federation ID
242        let client_config = fedimint_client::Client::get_config_from_db(&client_db)
243            .await
244            .context("Client config not found in database")?;
245
246        let federation_id = client_config.calculate_federation_id();
247
248        // Derive federation-specific secret from wallet mnemonic
249        let federation_secret = self.derive_federation_secret(&mnemonic, &federation_id);
250
251        let builder = Self::client_builder(client_db).await?;
252        let client = Arc::new(
253            builder
254                .open(RootSecret::StandardDoubleDerive(federation_secret))
255                .await?,
256        );
257
258        self.add_client(client_name, client).await;
259        Ok(())
260    }
261
262    async fn handle_close_client(&self, client_name: String) -> anyhow::Result<()> {
263        let mut clients = self.clients.lock().await;
264        let mut client = clients.remove(&client_name).context("client not found")?;
265
266        // RPC calls might have cloned the client Arc before we remove the client.
267        for attempt in 0.. {
268            info!(attempt, "waiting for RPCs to drop the federation object");
269            match Arc::try_unwrap(client) {
270                Ok(client) => {
271                    client.shutdown().await;
272                    break;
273                }
274                Err(client_val) => client = client_val,
275            }
276            fedimint_core::task::sleep(Duration::from_millis(100)).await;
277        }
278        Ok(())
279    }
280
281    fn handle_client_rpc(
282        self: Arc<Self>,
283        client_name: String,
284        module: String,
285        method: String,
286        payload: serde_json::Value,
287    ) -> BoxStream<'static, anyhow::Result<serde_json::Value>> {
288        Box::pin(try_stream! {
289            let client = self
290                .get_client(&client_name)
291                .await
292                .with_context(|| format!("Client not found: {client_name}"))?;
293            match module.as_str() {
294                "" => {
295                    let mut stream = client.handle_global_rpc(method, payload);
296                    while let Some(item) = stream.next().await {
297                        yield item?;
298                    }
299                }
300                "ln" => {
301                    let ln = client.get_first_module::<LightningClientModule>()?.inner();
302                    let mut stream = ln.handle_rpc(method, payload).await;
303                    while let Some(item) = stream.next().await {
304                        yield item?;
305                    }
306                }
307                "mint" => {
308                    let mint = client.get_first_module::<MintClientModule>()?.inner();
309                    let mut stream = mint.handle_rpc(method, payload).await;
310                    while let Some(item) = stream.next().await {
311                        yield item?;
312                    }
313                }
314                "wallet" => {
315                    let wallet = client
316                        .get_first_module::<WalletClientModule>()?
317                        .inner();
318                    let mut stream = wallet.handle_rpc(method, payload).await;
319                    while let Some(item) = stream.next().await {
320                        yield item?;
321                    }
322                }
323                _ => {
324                    Err(anyhow::format_err!("module not found: {module}"))?;
325                },
326            };
327        })
328    }
329
330    fn parse_invite_code(&self, invite_code: String) -> anyhow::Result<serde_json::Value> {
331        let invite_code = InviteCode::from_str(&invite_code)?;
332
333        Ok(json!({
334            "url": invite_code.url(),
335            "federation_id": invite_code.federation_id(),
336        }))
337    }
338
339    fn parse_bolt11_invoice(&self, invoice_str: String) -> anyhow::Result<serde_json::Value> {
340        let invoice = lightning_invoice::Bolt11Invoice::from_str(&invoice_str)
341            .map_err(|e| anyhow::anyhow!("Failed to parse Lightning invoice: {}", e))?;
342
343        let amount_msat = invoice.amount_milli_satoshis().unwrap_or(0);
344        let amount_sat = amount_msat as f64 / 1000.0;
345
346        let expiry_seconds = invoice.expiry_time().as_secs();
347
348        // memo
349        let description = match invoice.description() {
350            Bolt11InvoiceDescriptionRef::Direct(desc) => desc.to_string(),
351            Bolt11InvoiceDescriptionRef::Hash(_) => "Description hash only".to_string(),
352        };
353
354        Ok(json!({
355            "amount": amount_sat,
356            "expiry": expiry_seconds,
357            "memo": description,
358        }))
359    }
360
361    async fn preview_federation(&self, invite_code: String) -> anyhow::Result<serde_json::Value> {
362        let invite = InviteCode::from_str(&invite_code)?;
363        let client_config = fedimint_api_client::api::net::Connector::default()
364            .download_from_invite_code(&invite)
365            .await?;
366        let json_config = client_config.to_json();
367        let federation_id = client_config.calculate_federation_id();
368
369        Ok(json!({
370            "config": json_config,
371            "federation_id": federation_id.to_string(),
372        }))
373    }
374
375    fn handle_rpc_inner(
376        self: Arc<Self>,
377        request: RpcRequest,
378    ) -> Option<BoxStream<'static, anyhow::Result<serde_json::Value>>> {
379        match request.kind {
380            RpcRequestKind::SetMnemonic { words } => Some(Box::pin(try_stream! {
381                self.set_mnemonic(words).await?;
382                yield serde_json::json!({ "success": true });
383            })),
384            RpcRequestKind::GenerateMnemonic => Some(Box::pin(try_stream! {
385                let words = self.generate_mnemonic().await?;
386                yield serde_json::json!({ "mnemonic": words });
387            })),
388            RpcRequestKind::GetMnemonic => Some(Box::pin(try_stream! {
389                let words = self.get_mnemonic_words().await?;
390                yield serde_json::json!({ "mnemonic": words });
391            })),
392            RpcRequestKind::JoinFederation {
393                invite_code,
394                client_name,
395                force_recover,
396            } => Some(Box::pin(try_stream! {
397                self.handle_join_federation(invite_code, client_name, force_recover)
398                    .await?;
399                yield serde_json::json!(null);
400            })),
401            RpcRequestKind::OpenClient { client_name } => Some(Box::pin(try_stream! {
402                self.handle_open_client(client_name).await?;
403                yield serde_json::json!(null);
404            })),
405            RpcRequestKind::CloseClient { client_name } => Some(Box::pin(try_stream! {
406                self.handle_close_client(client_name).await?;
407                yield serde_json::json!(null);
408            })),
409            RpcRequestKind::ClientRpc {
410                client_name,
411                module,
412                method,
413                payload,
414            } => Some(self.handle_client_rpc(client_name, module, method, payload)),
415            RpcRequestKind::ParseInviteCode { invite_code } => Some(Box::pin(try_stream! {
416                let result = self.parse_invite_code(invite_code)?;
417                yield result;
418            })),
419            RpcRequestKind::ParseBolt11Invoice { invoice } => Some(Box::pin(try_stream! {
420                let result = self.parse_bolt11_invoice(invoice)?;
421                yield result;
422            })),
423            RpcRequestKind::PreviewFederation { invite_code } => Some(Box::pin(try_stream! {
424                let result = self.preview_federation(invite_code).await?;
425                yield result;
426            })),
427            RpcRequestKind::CancelRpc { cancel_request_id } => {
428                if let Some(handle) = self.remove_rpc_handle(cancel_request_id) {
429                    handle.abort();
430                }
431                None
432            }
433        }
434    }
435
436    pub fn handle_rpc(
437        self: Arc<Self>,
438        request: RpcRequest,
439        handler: impl RpcResponseHandler + 'static,
440    ) -> HandledRpc<'static> {
441        let request_id = request.request_id;
442
443        let Some(stream) = self.clone().handle_rpc_inner(request) else {
444            return HandledRpc { task: None };
445        };
446
447        let (abort_handle, abort_registration) = AbortHandle::new_pair();
448        self.add_rpc_handle(request_id, abort_handle);
449
450        let task = Box::pin(async move {
451            let mut stream = Abortable::new(stream, abort_registration);
452
453            while let Some(result) = stream.next().await {
454                let response = match result {
455                    Ok(value) => RpcResponse {
456                        request_id,
457                        kind: RpcResponseKind::Data { data: value },
458                    },
459                    Err(e) => RpcResponse {
460                        request_id,
461                        kind: RpcResponseKind::Error {
462                            error: e.to_string(),
463                        },
464                    },
465                };
466                handler.handle_response(response);
467            }
468
469            // Clean up abort handle and send end message
470            let _ = self.remove_rpc_handle(request_id);
471            handler.handle_response(RpcResponse {
472                request_id,
473                kind: if stream.is_aborted() {
474                    RpcResponseKind::Aborted {}
475                } else {
476                    RpcResponseKind::End {}
477                },
478            });
479        });
480
481        HandledRpc { task: Some(task) }
482    }
483
484    /// Retrieve the wallet-level mnemonic words.
485    /// Returns the mnemonic as a vector of words, or None if no mnemonic is
486    /// set.
487    async fn get_mnemonic_words(&self) -> anyhow::Result<Option<Vec<String>>> {
488        let mnemonic = self.get_mnemonic_from_db().await?;
489
490        if let Some(mnemonic) = mnemonic {
491            let words = mnemonic.words().map(|w| w.to_string()).collect();
492            Ok(Some(words))
493        } else {
494            Ok(None)
495        }
496    }
497    /// Set a mnemonic from user-provided words
498    /// Returns an error if a mnemonic is already set
499    async fn set_mnemonic(&self, words: Vec<String>) -> anyhow::Result<()> {
500        let all_words = words.join(" ");
501        let mnemonic =
502            Mnemonic::parse_in_normalized(fedimint_bip39::Language::English, &all_words)?;
503
504        let mut dbtx = self.unified_database.begin_transaction().await;
505
506        if dbtx.get_value(&MnemonicKey).await.is_some() {
507            anyhow::bail!(
508                "Wallet mnemonic already exists. Please clear existing data before setting a new mnemonic."
509            );
510        }
511
512        dbtx.insert_new_entry(&MnemonicKey, &mnemonic.to_entropy())
513            .await;
514
515        dbtx.commit_tx().await;
516
517        Ok(())
518    }
519
520    /// Generate a new random mnemonic and set it
521    /// Returns an error if a mnemonic is already set
522    async fn generate_mnemonic(&self) -> anyhow::Result<Vec<String>> {
523        let mnemonic = Bip39RootSecretStrategy::<12>::random(&mut thread_rng());
524        let words: Vec<String> = mnemonic.words().map(|w| w.to_string()).collect();
525
526        let mut dbtx = self.unified_database.begin_transaction().await;
527
528        if dbtx.get_value(&MnemonicKey).await.is_some() {
529            anyhow::bail!(
530                "Wallet mnemonic already exists. Please clear existing data before generating a new mnemonic."
531            );
532        }
533
534        dbtx.insert_new_entry(&MnemonicKey, &mnemonic.to_entropy())
535            .await;
536
537        dbtx.commit_tx().await;
538
539        Ok(words)
540    }
541
542    /// Derive federation-specific secret from wallet mnemonic
543    fn derive_federation_secret(
544        &self,
545        mnemonic: &Mnemonic,
546        federation_id: &FederationId,
547    ) -> DerivableSecret {
548        let global_root_secret = Bip39RootSecretStrategy::<12>::to_root_secret(mnemonic);
549        let multi_federation_root_secret = global_root_secret.child_key(ChildId(0));
550        let federation_root_secret = multi_federation_root_secret.federation_key(federation_id);
551        let federation_wallet_root_secret = federation_root_secret.child_key(ChildId(0));
552        federation_wallet_root_secret.child_key(ChildId(0))
553    }
554
555    /// Fetch mnemonic from database
556    async fn get_mnemonic_from_db(&self) -> anyhow::Result<Option<Mnemonic>> {
557        let mut dbtx = self.unified_database.begin_transaction_nc().await;
558
559        if let Some(mnemonic_entropy) = dbtx.get_value(&MnemonicKey).await {
560            let mnemonic = Mnemonic::from_entropy(&mnemonic_entropy)?;
561            Ok(Some(mnemonic))
562        } else {
563            Ok(None)
564        }
565    }
566}
fedimint_client_rpc/lib.rs

fedimint_client_rpc/
lib.rs
