1use std::collections::HashMap;
2use std::str::FromStr;
3use std::sync::Arc;
4use std::time::Duration;
5
6use async_stream::try_stream;
7use fedimint_api_client::api::{ClientConfigDownloadError, FederationError};
8use fedimint_bip39::{Bip39RootSecretStrategy, Mnemonic};
9use fedimint_client::error::{
10 ClientBuildError, ClientModuleError, GlobalRpcError, ModuleLookupError,
11};
12use fedimint_client::module::ClientModule;
13use fedimint_client::secret::RootSecretStrategy;
14use fedimint_client::{ClientHandleArc, ClientPreview, RootSecret};
15use fedimint_connectors::ConnectorRegistry;
16use fedimint_core::config::{FederationId, FederationIdPrefix};
17use fedimint_core::db::{Database, IDatabaseTransactionOpsCoreTyped};
18use fedimint_core::encoding::{Decodable, Encodable};
19use fedimint_core::invite_code::{InviteCode, InviteCodeParseError};
20use fedimint_core::task::{MaybeSend, MaybeSync};
21use fedimint_core::util::{BoxFuture, BoxStream, FmtCompact as _};
22use fedimint_core::{Amount, TieredCounts, impl_db_record};
23use fedimint_derive_secret::{ChildId, DerivableSecret};
24use fedimint_ln_client::{LightningClientInit, LightningClientModule};
25use fedimint_meta_client::{MetaClientInit, MetaClientModule};
26use fedimint_mint_client::{MintClientInit, MintClientModule, OOBNotes, OOBNotesParseError};
27use fedimint_wallet_client::{WalletClientInit, WalletClientModule};
28use futures::StreamExt;
29use futures::future::{AbortHandle, Abortable};
30use lightning_invoice::{Bolt11InvoiceDescriptionRef, ParseOrSemanticError};
31use rand::thread_rng;
32use serde::{Deserialize, Serialize};
33use serde_json::json;
34use tokio::sync::Mutex;
35use tracing::info;
36
37#[repr(u8)]
39#[derive(Clone, Copy, Debug)]
40pub enum DbKeyPrefix {
41 ClientDatabase = 0x00,
42 Mnemonic = 0x01,
43}
44
45#[derive(Debug, Clone, Encodable, Decodable, Eq, PartialEq, Hash)]
46pub struct MnemonicKey;
47
48impl_db_record!(
49 key = MnemonicKey,
50 value = Vec<u8>,
51 db_prefix = DbKeyPrefix::Mnemonic,
52);
53
54#[derive(Debug, Clone, Serialize, Deserialize)]
56pub struct ParsedNoteDetails {
57 pub total_amount: Amount,
59 pub federation_id_prefix: FederationIdPrefix,
61 pub federation_id: Option<FederationId>,
63 pub invite_code: Option<InviteCode>,
65 pub note_counts: TieredCounts,
67}
68
69#[derive(Serialize, Deserialize)]
70#[serde(rename_all = "snake_case")]
71pub struct RpcRequest {
72 pub request_id: u64,
73 #[serde(flatten)]
74 pub kind: RpcRequestKind,
75}
76
77#[derive(Serialize, Deserialize)]
78#[serde(tag = "type", rename_all = "snake_case")]
79pub enum RpcRequestKind {
80 SetMnemonic {
81 words: Vec<String>,
82 },
83 GenerateMnemonic,
84 GetMnemonic,
85 HasMnemonicSet,
86 JoinFederation {
88 invite_code: String,
89 force_recover: bool,
90 client_name: String,
91 },
92 OpenClient {
93 client_name: String,
94 },
95 CloseClient {
96 client_name: String,
97 },
98 ClientRpc {
99 client_name: String,
100 module: String,
101 method: String,
102 payload: serde_json::Value,
103 },
104 CancelRpc {
105 cancel_request_id: u64,
106 },
107 ParseInviteCode {
108 invite_code: String,
109 },
110 ParseBolt11Invoice {
111 invoice: String,
112 },
113 PreviewFederation {
114 invite_code: String,
115 },
116 ParseOobNotes {
117 oob_notes: String,
118 },
119 ParseLightningAddress {
120 address: String,
121 },
122}
123
124#[derive(Serialize, Deserialize, Clone, Debug)]
125pub struct RpcResponse {
126 pub request_id: u64,
127 #[serde(flatten)]
128 pub kind: RpcResponseKind,
129}
130
131#[derive(Serialize, Deserialize, Clone, Debug)]
132#[serde(tag = "type", rename_all = "snake_case")]
133pub enum RpcResponseKind {
134 Data { data: serde_json::Value },
135 Error { error: String },
136 Aborted {},
137 End {},
138}
139
140pub trait RpcResponseHandler: MaybeSend + MaybeSync {
141 fn handle_response(&self, response: RpcResponse);
142}
143
144pub struct RpcGlobalState {
145 connectors: ConnectorRegistry,
147 clients: Mutex<HashMap<String, ClientHandleArc>>,
148 rpc_handles: std::sync::Mutex<HashMap<u64, AbortHandle>>,
149 unified_database: Database,
150 preview_cache: std::sync::Mutex<Option<ClientPreview>>,
151}
152
153pub struct HandledRpc<'a> {
154 pub task: Option<BoxFuture<'a, ()>>,
155}
156
157impl RpcGlobalState {
158 pub fn new(connectors: ConnectorRegistry, unified_database: Database) -> Self {
159 Self {
160 connectors,
161 clients: Mutex::new(HashMap::new()),
162 rpc_handles: std::sync::Mutex::new(HashMap::new()),
163 unified_database,
164 preview_cache: std::sync::Mutex::new(None),
165 }
166 }
167
168 async fn add_client(&self, client_name: String, client: ClientHandleArc) {
169 let mut clients = self.clients.lock().await;
170 clients.insert(client_name, client);
171 }
172
173 async fn get_client(&self, client_name: &str) -> Option<ClientHandleArc> {
174 let clients = self.clients.lock().await;
175 clients.get(client_name).cloned()
176 }
177
178 fn add_rpc_handle(&self, request_id: u64, handle: AbortHandle) {
179 let mut handles = self.rpc_handles.lock().unwrap();
180 if handles.insert(request_id, handle).is_some() {
181 tracing::error!("RPC CLIENT ERROR: request id reuse detected");
182 }
183 }
184
185 fn remove_rpc_handle(&self, request_id: u64) -> Option<AbortHandle> {
186 let mut handles = self.rpc_handles.lock().unwrap();
187 handles.remove(&request_id)
188 }
189
190 async fn client_builder() -> fedimint_client::ClientBuilder {
191 let mut builder = fedimint_client::Client::builder().await;
192 builder.with_module(MintClientInit);
193 builder.with_module(LightningClientInit::default());
194 builder.with_module(WalletClientInit(None));
195 builder.with_module(MetaClientInit);
196 builder
197 }
198
199 async fn client_db(&self, client_name: String) -> Database {
201 assert_eq!(client_name.len(), 36);
202
203 let unified_db = &self.unified_database;
204 let mut client_prefix = vec![DbKeyPrefix::ClientDatabase as u8];
205 client_prefix.extend_from_slice(client_name.as_bytes());
206 unified_db.with_prefix(client_prefix)
207 }
208
209 async fn handle_join_federation(
211 &self,
212
213 invite_code: String,
214 client_name: String,
215 force_recover: bool,
216 ) -> Result<(), RpcRequestError> {
217 let mnemonic = self
219 .get_mnemonic_from_db()
220 .await?
221 .ok_or(RpcRequestError::NoMnemonic)?;
222
223 let client_db = self.client_db(client_name.clone()).await;
224
225 let invite_code = InviteCode::from_str(&invite_code)?;
226 let federation_id = invite_code.federation_id();
227
228 let federation_secret = self.derive_federation_secret(&mnemonic, &federation_id);
230
231 let cached_preview = self.preview_cache.lock().unwrap().take();
233 let preview = match cached_preview {
234 Some(preview) if preview.config().calculate_federation_id() == federation_id => preview,
235 _ => {
236 let builder = Self::client_builder().await;
237 builder
238 .preview(self.connectors.clone(), &invite_code)
239 .await?
240 }
241 };
242
243 #[allow(deprecated)]
245 let backup = preview
246 .download_backup_from_federation(RootSecret::StandardDoubleDerive(
247 federation_secret.clone(),
248 ))
249 .await?;
250
251 let client = if force_recover || backup.is_some() {
252 Arc::new(
253 preview
254 .recover(
255 client_db,
256 RootSecret::StandardDoubleDerive(federation_secret),
257 backup,
258 )
259 .await?,
260 )
261 } else {
262 Arc::new(
263 preview
264 .join(
265 client_db,
266 RootSecret::StandardDoubleDerive(federation_secret),
267 )
268 .await?,
269 )
270 };
271
272 self.add_client(client_name, client).await;
273 Ok(())
274 }
275
276 async fn handle_open_client(&self, client_name: String) -> Result<(), RpcRequestError> {
277 let mnemonic = self
279 .get_mnemonic_from_db()
280 .await?
281 .ok_or(RpcRequestError::NoMnemonic)?;
282
283 let client_db = self.client_db(client_name.clone()).await;
284
285 if !fedimint_client::Client::is_initialized(&client_db).await {
286 return Err(RpcRequestError::ClientNotInitialized);
287 }
288
289 let client_config = fedimint_client::Client::get_config_from_db(&client_db)
291 .await
292 .ok_or(RpcRequestError::NoClientConfig)?;
293
294 let federation_id = client_config.calculate_federation_id();
295
296 let federation_secret = self.derive_federation_secret(&mnemonic, &federation_id);
298
299 let builder = Self::client_builder().await;
300 let client = Arc::new(
301 builder
302 .open(
303 self.connectors.clone(),
304 client_db,
305 RootSecret::StandardDoubleDerive(federation_secret),
306 )
307 .await?,
308 );
309
310 self.add_client(client_name, client).await;
311 Ok(())
312 }
313
314 async fn handle_close_client(&self, client_name: String) -> Result<(), RpcRequestError> {
315 let mut clients = self.clients.lock().await;
316 let mut client = clients
317 .remove(&client_name)
318 .ok_or(RpcRequestError::ClientToCloseNotFound)?;
319
320 for attempt in 0.. {
322 info!(attempt, "waiting for RPCs to drop the federation object");
323 match Arc::try_unwrap(client) {
324 Ok(client) => {
325 client.shutdown().await;
326 break;
327 }
328 Err(client_val) => client = client_val,
329 }
330 fedimint_core::task::sleep(Duration::from_millis(100)).await;
331 }
332 Ok(())
333 }
334
335 fn handle_client_rpc(
336 self: Arc<Self>,
337 client_name: String,
338 module: String,
339 method: String,
340 payload: serde_json::Value,
341 ) -> BoxStream<'static, Result<serde_json::Value, RpcRequestError>> {
342 Box::pin(try_stream! {
343 let client = self
344 .get_client(&client_name)
345 .await
346 .ok_or_else(|| RpcRequestError::ClientNotFound {
347 client_name: client_name.clone(),
348 })?;
349 match module.as_str() {
350 "" => {
351 let mut stream = client.handle_global_rpc(method, payload);
352 while let Some(item) = stream.next().await {
353 yield item?;
354 }
355 }
356 "ln" => {
357 let ln = client.get_first_module::<LightningClientModule>()?.inner();
358 let mut stream = ln.handle_rpc(method, payload).await;
359 while let Some(item) = stream.next().await {
360 yield item?;
361 }
362 }
363 "mint" => {
364 let mint = client.get_first_module::<MintClientModule>()?.inner();
365 let mut stream = mint.handle_rpc(method, payload).await;
366 while let Some(item) = stream.next().await {
367 yield item?;
368 }
369 }
370 "wallet" => {
371 let wallet = client
372 .get_first_module::<WalletClientModule>()?
373 .inner();
374 let mut stream = wallet.handle_rpc(method, payload).await;
375 while let Some(item) = stream.next().await {
376 yield item?;
377 }
378 }
379 "meta" => {
380 let meta = client.get_first_module::<MetaClientModule>()?.inner();
381 let mut stream = meta.handle_rpc(method, payload).await;
382 while let Some(item) = stream.next().await {
383 yield item?;
384 }
385 }
386 _ => {
387 Err(RpcRequestError::UnknownModule { module: module.clone() })?;
388 },
389 };
390 })
391 }
392
393 fn parse_invite_code(
394 &self,
395 invite_code: String,
396 ) -> Result<serde_json::Value, InviteCodeParseError> {
397 let invite_code = InviteCode::from_str(&invite_code)?;
398
399 Ok(json!({
400 "url": invite_code.url(),
401 "federation_id": invite_code.federation_id(),
402 }))
403 }
404
405 fn parse_bolt11_invoice(
406 &self,
407 invoice_str: String,
408 ) -> Result<serde_json::Value, RpcRequestError> {
409 let invoice = lightning_invoice::Bolt11Invoice::from_str(&invoice_str)
410 .map_err(RpcRequestError::InvalidInvoice)?;
411
412 let amount_msat = invoice.amount_milli_satoshis().unwrap_or(0);
413 let amount_sat = amount_msat as f64 / 1000.0;
414
415 let expiry_seconds = invoice.expiry_time().as_secs();
416
417 let description = match invoice.description() {
419 Bolt11InvoiceDescriptionRef::Direct(desc) => desc.to_string(),
420 Bolt11InvoiceDescriptionRef::Hash(_) => "Description hash only".to_string(),
421 };
422
423 Ok(json!({
424 "amount": amount_sat,
425 "expiry": expiry_seconds,
426 "memo": description,
427 }))
428 }
429
430 async fn preview_federation(
431 &self,
432 invite_code: String,
433 ) -> Result<serde_json::Value, RpcRequestError> {
434 let invite = InviteCode::from_str(&invite_code)?;
435 let federation_id = invite.federation_id();
436
437 let builder = Self::client_builder().await;
438 let preview = builder.preview(self.connectors.clone(), &invite).await?;
439
440 let json_config = preview.config().to_json();
441 *self.preview_cache.lock().unwrap() = Some(preview);
443
444 Ok(json!({
445 "config": json_config,
446 "federation_id": federation_id.to_string(),
447 }))
448 }
449
450 fn handle_rpc_inner(
451 self: Arc<Self>,
452 request: RpcRequest,
453 ) -> Option<BoxStream<'static, Result<serde_json::Value, RpcRequestError>>> {
454 match request.kind {
455 RpcRequestKind::SetMnemonic { words } => Some(Box::pin(try_stream! {
456 self.set_mnemonic(words).await?;
457 yield serde_json::json!({ "success": true });
458 })),
459 RpcRequestKind::GenerateMnemonic => Some(Box::pin(try_stream! {
460 let words = self.generate_mnemonic().await?;
461 yield serde_json::json!({ "mnemonic": words });
462 })),
463 RpcRequestKind::GetMnemonic => Some(Box::pin(try_stream! {
464 let words = self.get_mnemonic_words().await?;
465 yield serde_json::json!({ "mnemonic": words });
466 })),
467 RpcRequestKind::HasMnemonicSet => Some(Box::pin(try_stream! {
468 let is_set = self.has_mnemonic_set().await?;
469 yield serde_json::json!(is_set);
470 })),
471 RpcRequestKind::JoinFederation {
472 invite_code,
473 client_name,
474 force_recover,
475 } => Some(Box::pin(try_stream! {
476 self.handle_join_federation(invite_code, client_name, force_recover)
477 .await?;
478 yield serde_json::json!(null);
479 })),
480 RpcRequestKind::OpenClient { client_name } => Some(Box::pin(try_stream! {
481 self.handle_open_client(client_name).await?;
482 yield serde_json::json!(null);
483 })),
484 RpcRequestKind::CloseClient { client_name } => Some(Box::pin(try_stream! {
485 self.handle_close_client(client_name).await?;
486 yield serde_json::json!(null);
487 })),
488 RpcRequestKind::ClientRpc {
489 client_name,
490 module,
491 method,
492 payload,
493 } => Some(self.handle_client_rpc(client_name, module, method, payload)),
494 RpcRequestKind::ParseInviteCode { invite_code } => Some(Box::pin(try_stream! {
495 let result = self.parse_invite_code(invite_code)?;
496 yield result;
497 })),
498 RpcRequestKind::ParseBolt11Invoice { invoice } => Some(Box::pin(try_stream! {
499 let result = self.parse_bolt11_invoice(invoice)?;
500 yield result;
501 })),
502 RpcRequestKind::PreviewFederation { invite_code } => Some(Box::pin(try_stream! {
503 let result = self.preview_federation(invite_code).await?;
504 yield result;
505 })),
506 RpcRequestKind::ParseOobNotes { oob_notes } => Some(Box::pin(try_stream! {
507 let parsed = parse_oob_notes(&oob_notes)?;
508 yield serde_json::to_value(parsed)?;
509 })),
510 RpcRequestKind::ParseLightningAddress { address } => Some(Box::pin(try_stream! {
511 let url = fedimint_lnurl::parse_address(&address)
512 .ok_or(RpcRequestError::InvalidLightningAddress)?;
513 let metadata = fedimint_lnurl::request(&url).await
514 .map_err(RpcRequestError::LnurlRequest)?;
515
516 yield serde_json::to_value(metadata)?;
517 })),
518 RpcRequestKind::CancelRpc { cancel_request_id } => {
519 if let Some(handle) = self.remove_rpc_handle(cancel_request_id) {
520 handle.abort();
521 }
522 None
523 }
524 }
525 }
526
527 pub fn handle_rpc(
528 self: Arc<Self>,
529 request: RpcRequest,
530 handler: impl RpcResponseHandler + 'static,
531 ) -> HandledRpc<'static> {
532 let request_id = request.request_id;
533
534 let Some(stream) = self.clone().handle_rpc_inner(request) else {
535 return HandledRpc { task: None };
536 };
537
538 let (abort_handle, abort_registration) = AbortHandle::new_pair();
539 self.add_rpc_handle(request_id, abort_handle);
540
541 let task = Box::pin(async move {
542 let mut stream = Abortable::new(stream, abort_registration);
543
544 while let Some(result) = stream.next().await {
545 let response = match result {
546 Ok(value) => RpcResponse {
547 request_id,
548 kind: RpcResponseKind::Data { data: value },
549 },
550 Err(e) => RpcResponse {
551 request_id,
552 kind: RpcResponseKind::Error {
553 error: e.fmt_compact().to_string(),
554 },
555 },
556 };
557 handler.handle_response(response);
558 }
559
560 let _ = self.remove_rpc_handle(request_id);
562 handler.handle_response(RpcResponse {
563 request_id,
564 kind: if stream.is_aborted() {
565 RpcResponseKind::Aborted {}
566 } else {
567 RpcResponseKind::End {}
568 },
569 });
570 });
571
572 HandledRpc { task: Some(task) }
573 }
574
575 async fn get_mnemonic_words(&self) -> Result<Option<Vec<String>>, bip39::Error> {
579 let mnemonic = self.get_mnemonic_from_db().await?;
580
581 if let Some(mnemonic) = mnemonic {
582 let words = mnemonic.words().map(|w| w.to_string()).collect();
583 Ok(Some(words))
584 } else {
585 Ok(None)
586 }
587 }
588 async fn set_mnemonic(&self, words: Vec<String>) -> Result<(), RpcRequestError> {
591 let all_words = words.join(" ");
592 let mnemonic =
593 Mnemonic::parse_in_normalized(fedimint_bip39::Language::English, &all_words)?;
594
595 let mut dbtx = self.unified_database.begin_transaction().await;
596
597 if dbtx.get_value(&MnemonicKey).await.is_some() {
598 return Err(RpcRequestError::MnemonicExistsOnSet);
599 }
600
601 dbtx.insert_new_entry(&MnemonicKey, &mnemonic.to_entropy())
602 .await;
603
604 dbtx.commit_tx().await;
605
606 Ok(())
607 }
608
609 async fn generate_mnemonic(&self) -> Result<Vec<String>, RpcRequestError> {
612 let mnemonic = Bip39RootSecretStrategy::<12>::random(&mut thread_rng());
613 let words: Vec<String> = mnemonic.words().map(|w| w.to_string()).collect();
614
615 let mut dbtx = self.unified_database.begin_transaction().await;
616
617 if dbtx.get_value(&MnemonicKey).await.is_some() {
618 return Err(RpcRequestError::MnemonicExistsOnGenerate);
619 }
620
621 dbtx.insert_new_entry(&MnemonicKey, &mnemonic.to_entropy())
622 .await;
623
624 dbtx.commit_tx().await;
625
626 Ok(words)
627 }
628
629 fn derive_federation_secret(
631 &self,
632 mnemonic: &Mnemonic,
633 federation_id: &FederationId,
634 ) -> DerivableSecret {
635 let global_root_secret = Bip39RootSecretStrategy::<12>::to_root_secret(mnemonic);
636 let multi_federation_root_secret = global_root_secret.child_key(ChildId(0));
637 let federation_root_secret = multi_federation_root_secret.federation_key(federation_id);
638 let federation_wallet_root_secret = federation_root_secret.child_key(ChildId(0));
639 federation_wallet_root_secret.child_key(ChildId(0))
640 }
641
642 async fn get_mnemonic_from_db(&self) -> Result<Option<Mnemonic>, bip39::Error> {
644 let mut dbtx = self.unified_database.begin_transaction_nc().await;
645
646 if let Some(mnemonic_entropy) = dbtx.get_value(&MnemonicKey).await {
647 let mnemonic = Mnemonic::from_entropy(&mnemonic_entropy)?;
648 Ok(Some(mnemonic))
649 } else {
650 Ok(None)
651 }
652 }
653
654 async fn has_mnemonic_set(&self) -> Result<bool, bip39::Error> {
656 let mnemonic = self.get_mnemonic_from_db().await?;
657 Ok(mnemonic.is_some())
658 }
659}
660
661pub fn parse_oob_notes(oob_notes_str: &str) -> Result<ParsedNoteDetails, OOBNotesParseError> {
662 let oob_notes = OOBNotes::from_str(oob_notes_str)?;
663
664 let total_amount = oob_notes.total_amount();
665 let federation_id_prefix = oob_notes.federation_id_prefix();
666 let invite_code = oob_notes.federation_invite();
667 let federation_id = invite_code.as_ref().map(|inv| inv.federation_id());
668
669 let notes = oob_notes.notes();
671 let mut note_counts = TieredCounts::default();
672 for (amount, _note) in notes.iter_items() {
673 note_counts.inc(amount, 1);
674 }
675
676 Ok(ParsedNoteDetails {
677 total_amount,
678 federation_id_prefix,
679 federation_id,
680 invite_code,
681 note_counts,
682 })
683}
684
685#[derive(Debug, thiserror::Error)]
688enum RpcRequestError {
689 #[error("No wallet mnemonic set. Please set or generate a mnemonic first.")]
691 NoMnemonic,
692
693 #[error(
695 "Wallet mnemonic already exists. Please clear existing data before setting a new \
696 mnemonic."
697 )]
698 MnemonicExistsOnSet,
699
700 #[error(
702 "Wallet mnemonic already exists. Please clear existing data before generating a new \
703 mnemonic."
704 )]
705 MnemonicExistsOnGenerate,
706
707 #[error(transparent)]
710 Mnemonic(#[from] bip39::Error),
711
712 #[error(transparent)]
714 InviteCode(#[from] InviteCodeParseError),
715
716 #[error(transparent)]
719 ConfigDownload(#[from] ClientConfigDownloadError),
720
721 #[error(transparent)]
723 Federation(#[from] FederationError),
724
725 #[error(transparent)]
727 ClientBuild(#[from] ClientBuildError),
728
729 #[error("client is not initialized for this database")]
731 ClientNotInitialized,
732
733 #[error("Client config not found in database")]
735 NoClientConfig,
736
737 #[error("client not found")]
739 ClientToCloseNotFound,
740
741 #[error("Client not found: {client_name}")]
743 ClientNotFound { client_name: String },
744
745 #[error("module not found: {module}")]
747 UnknownModule { module: String },
748
749 #[error(transparent)]
751 ModuleLookup(#[from] ModuleLookupError),
752
753 #[error(transparent)]
755 GlobalRpc(#[from] GlobalRpcError),
756
757 #[error(transparent)]
759 Module(#[from] ClientModuleError),
760
761 #[error("Failed to parse Lightning invoice")]
763 InvalidInvoice(#[source] ParseOrSemanticError),
764
765 #[error(transparent)]
767 OobNotes(#[from] OOBNotesParseError),
768
769 #[error("Invalid Lightning Address")]
771 InvalidLightningAddress,
772
773 #[error("{0}")]
776 LnurlRequest(String),
777
778 #[error(transparent)]
780 Json(#[from] serde_json::Error),
781}
782
783#[cfg(test)]
784mod tests;