cdk_redb/wallet/
mod.rs

1//! Redb Wallet
2
3use std::cmp::Ordering;
4use std::collections::HashMap;
5use std::path::Path;
6use std::str::FromStr;
7use std::sync::Arc;
8
9use async_trait::async_trait;
10use cdk_common::bitcoin::bip32::DerivationPath;
11use cdk_common::database::{validate_kvstore_params, WalletDatabase};
12use cdk_common::mint_url::MintUrl;
13use cdk_common::nut00::KnownMethod;
14use cdk_common::util::unix_time;
15use cdk_common::wallet::{
16    self, MintQuote, ProofInfo, Transaction, TransactionDirection, TransactionId,
17};
18use cdk_common::{
19    database, Amount, CurrencyUnit, Id, KeySet, KeySetInfo, Keys, MintInfo, PaymentMethod,
20    PublicKey, SpendingConditions, State,
21};
22use redb::{Database, MultimapTableDefinition, ReadableDatabase, ReadableTable, TableDefinition};
23use tracing::instrument;
24
25use crate::error::Error;
26use crate::migrations::migrate_00_to_01;
27use crate::wallet::migrations::{
28    migrate_01_to_02, migrate_02_to_03, migrate_03_to_04, migrate_04_to_05, migrate_05_to_06,
29};
30
31mod migrations;
32
33// <Mint_url, Info>
34const MINTS_TABLE: TableDefinition<&str, &str> = TableDefinition::new("mints_table");
35// <Mint_Url, Keyset_id>
36const MINT_KEYSETS_TABLE: MultimapTableDefinition<&str, &[u8]> =
37    MultimapTableDefinition::new("mint_keysets");
38// <Keyset_id, KeysetInfo>
39const KEYSETS_TABLE: TableDefinition<&[u8], &str> = TableDefinition::new("keysets");
40// <Quote_id, quote>
41const MINT_QUOTES_TABLE: TableDefinition<&str, &str> = TableDefinition::new("mint_quotes");
42// <Quote_id, quote>
43const MELT_QUOTES_TABLE: TableDefinition<&str, &str> = TableDefinition::new("melt_quotes");
44const MINT_KEYS_TABLE: TableDefinition<&str, &str> = TableDefinition::new("mint_keys");
45// <Y, Proof Info>
46const PROOFS_TABLE: TableDefinition<&[u8], &str> = TableDefinition::new("proofs");
47const CONFIG_TABLE: TableDefinition<&str, &str> = TableDefinition::new("config");
48const KEYSET_COUNTER: TableDefinition<&str, u32> = TableDefinition::new("keyset_counter");
49const DERIVATION_COUNTER: TableDefinition<&str, u32> = TableDefinition::new("derivation_counter");
50// <Transaction_id, Transaction>
51const TRANSACTIONS_TABLE: TableDefinition<&[u8], &str> = TableDefinition::new("transactions");
52// <Saga_id, WalletSaga>
53const SAGAS_TABLE: TableDefinition<&str, &str> = TableDefinition::new("wallet_sagas");
54
55// <Pubkey, P2PKSigningKey>
56const P2PK_SIGNING_KEYS_TABLE: TableDefinition<&[u8], &str> =
57    TableDefinition::new("p2pk_signing_keys");
58
59const KEYSET_U32_MAPPING: TableDefinition<u32, &str> = TableDefinition::new("keyset_u32_mapping");
60// <(primary_namespace, secondary_namespace, key), value>
61const KV_STORE_TABLE: TableDefinition<(&str, &str, &str), &[u8]> = TableDefinition::new("kv_store");
62
63const DATABASE_VERSION: u32 = 6;
64
65/// Wallet Redb Database
66#[derive(Debug, Clone)]
67pub struct WalletRedbDatabase {
68    db: Arc<Database>,
69}
70
71impl WalletRedbDatabase {
72    /// Create new [`WalletRedbDatabase`]
73    pub fn new(path: &Path) -> Result<Self, Error> {
74        {
75            // Check if parent directory exists before attempting to create database
76            if let Some(parent) = path.parent() {
77                if !parent.exists() {
78                    return Err(Error::Io(std::io::Error::new(
79                        std::io::ErrorKind::NotFound,
80                        format!("Parent directory does not exist: {parent:?}"),
81                    )));
82                }
83            }
84
85            let db = Arc::new(Database::create(path)?);
86
87            let db_version: Option<String>;
88            {
89                // Check database version
90                let read_txn = db.begin_read()?;
91                let table = read_txn.open_table(CONFIG_TABLE);
92
93                db_version = match table {
94                    Ok(table) => table.get("db_version")?.map(|v| v.value().to_string()),
95                    Err(_) => None,
96                };
97            }
98
99            match db_version {
100                Some(db_version) => {
101                    let mut current_file_version = u32::from_str(&db_version)?;
102                    tracing::info!("Current file version {}", current_file_version);
103
104                    match current_file_version.cmp(&DATABASE_VERSION) {
105                        Ordering::Less => {
106                            tracing::info!(
107                                "Database needs to be upgraded at {} current is {}",
108                                current_file_version,
109                                DATABASE_VERSION
110                            );
111                            if current_file_version == 0 {
112                                current_file_version = migrate_00_to_01(Arc::clone(&db))?;
113                            }
114
115                            if current_file_version == 1 {
116                                current_file_version = migrate_01_to_02(Arc::clone(&db))?;
117                            }
118
119                            if current_file_version == 2 {
120                                current_file_version = migrate_02_to_03(Arc::clone(&db))?;
121                            }
122
123                            if current_file_version == 3 {
124                                current_file_version = migrate_03_to_04(Arc::clone(&db))?;
125                            }
126
127                            if current_file_version == 4 {
128                                current_file_version = migrate_04_to_05(Arc::clone(&db))?;
129                            }
130
131                            if current_file_version == 5 {
132                                current_file_version = migrate_05_to_06(Arc::clone(&db))?;
133                            }
134
135                            if current_file_version != DATABASE_VERSION {
136                                tracing::warn!(
137                                    "Database upgrade did not complete at {} current is {}",
138                                    current_file_version,
139                                    DATABASE_VERSION
140                                );
141                                return Err(Error::UnknownDatabaseVersion);
142                            }
143
144                            let write_txn = db.begin_write()?;
145                            {
146                                let mut table = write_txn.open_table(CONFIG_TABLE)?;
147
148                                table
149                                    .insert("db_version", DATABASE_VERSION.to_string().as_str())?;
150                            }
151
152                            write_txn.commit()?;
153                        }
154                        Ordering::Equal => {
155                            tracing::info!("Database is at current version {}", DATABASE_VERSION);
156                        }
157                        Ordering::Greater => {
158                            tracing::warn!(
159                                "Database upgrade did not complete at {} current is {}",
160                                current_file_version,
161                                DATABASE_VERSION
162                            );
163                            return Err(Error::UnknownDatabaseVersion);
164                        }
165                    }
166                }
167                None => {
168                    let write_txn = db.begin_write()?;
169                    {
170                        let mut table = write_txn.open_table(CONFIG_TABLE)?;
171                        // Open all tables to init a new db
172                        let _ = write_txn.open_table(MINTS_TABLE)?;
173                        let _ = write_txn.open_multimap_table(MINT_KEYSETS_TABLE)?;
174                        let _ = write_txn.open_table(KEYSETS_TABLE)?;
175                        let _ = write_txn.open_table(MINT_QUOTES_TABLE)?;
176                        let _ = write_txn.open_table(MELT_QUOTES_TABLE)?;
177                        let _ = write_txn.open_table(MINT_KEYS_TABLE)?;
178                        let _ = write_txn.open_table(PROOFS_TABLE)?;
179                        let _ = write_txn.open_table(KEYSET_COUNTER)?;
180                        let _ = write_txn.open_table(TRANSACTIONS_TABLE)?;
181                        let _ = write_txn.open_table(KEYSET_U32_MAPPING)?;
182                        let _ = write_txn.open_table(KV_STORE_TABLE)?;
183                        let _ = write_txn.open_table(P2PK_SIGNING_KEYS_TABLE)?;
184                        table.insert("db_version", DATABASE_VERSION.to_string().as_str())?;
185                    }
186
187                    write_txn.commit()?;
188                }
189            }
190            drop(db);
191        }
192
193        // Check parent directory again for final database creation
194        if let Some(parent) = path.parent() {
195            if !parent.exists() {
196                return Err(Error::Io(std::io::Error::new(
197                    std::io::ErrorKind::NotFound,
198                    format!("Parent directory does not exist: {parent:?}"),
199                )));
200            }
201        }
202
203        let db = Database::create(path)?;
204
205        Ok(Self { db: Arc::new(db) })
206    }
207}
208
209#[async_trait]
210impl WalletDatabase<database::Error> for WalletRedbDatabase {
211    #[instrument(skip(self))]
212    async fn get_mint(&self, mint_url: MintUrl) -> Result<Option<MintInfo>, database::Error> {
213        let read_txn = self.db.begin_read().map_err(Into::<Error>::into)?;
214        let table = read_txn.open_table(MINTS_TABLE).map_err(Error::from)?;
215
216        if let Some(mint_info) = table
217            .get(mint_url.to_string().as_str())
218            .map_err(Error::from)?
219        {
220            return Ok(serde_json::from_str(mint_info.value()).map_err(Error::from)?);
221        }
222
223        Ok(None)
224    }
225
226    #[instrument(skip(self))]
227    async fn get_mints(&self) -> Result<HashMap<MintUrl, Option<MintInfo>>, database::Error> {
228        let read_txn = self.db.begin_read().map_err(Error::from)?;
229        let table = read_txn.open_table(MINTS_TABLE).map_err(Error::from)?;
230        let mints = table
231            .iter()
232            .map_err(Error::from)?
233            .flatten()
234            .filter_map(|(mint, mint_info)| {
235                MintUrl::from_str(mint.value())
236                    .ok()
237                    .map(|url| (url, serde_json::from_str(mint_info.value()).ok()))
238            })
239            .collect();
240
241        Ok(mints)
242    }
243
244    #[instrument(skip(self))]
245    async fn get_mint_keysets(
246        &self,
247        mint_url: MintUrl,
248    ) -> Result<Option<Vec<KeySetInfo>>, database::Error> {
249        let read_txn = self.db.begin_read().map_err(Into::<Error>::into)?;
250        let table = read_txn
251            .open_multimap_table(MINT_KEYSETS_TABLE)
252            .map_err(Error::from)?;
253
254        let keyset_ids = table
255            .get(mint_url.to_string().as_str())
256            .map_err(Error::from)?
257            .flatten()
258            .map(|k| Id::from_bytes(k.value()))
259            .collect::<Result<Vec<_>, _>>()?;
260
261        let mut keysets = vec![];
262
263        let keysets_t = read_txn.open_table(KEYSETS_TABLE).map_err(Error::from)?;
264
265        for keyset_id in keyset_ids {
266            if let Some(keyset) = keysets_t
267                .get(keyset_id.to_bytes().as_slice())
268                .map_err(Error::from)?
269            {
270                let keyset = serde_json::from_str(keyset.value()).map_err(Error::from)?;
271
272                keysets.push(keyset);
273            }
274        }
275
276        match keysets.is_empty() {
277            true => Ok(None),
278            false => Ok(Some(keysets)),
279        }
280    }
281
282    #[instrument(skip(self), fields(keyset_id = %keyset_id))]
283    async fn get_keyset_by_id(
284        &self,
285        keyset_id: &Id,
286    ) -> Result<Option<KeySetInfo>, database::Error> {
287        let read_txn = self.db.begin_read().map_err(Into::<Error>::into)?;
288        let table = read_txn.open_table(KEYSETS_TABLE).map_err(Error::from)?;
289
290        match table
291            .get(keyset_id.to_bytes().as_slice())
292            .map_err(Error::from)?
293        {
294            Some(keyset) => {
295                let keyset: KeySetInfo =
296                    serde_json::from_str(keyset.value()).map_err(Error::from)?;
297
298                Ok(Some(keyset))
299            }
300            None => Ok(None),
301        }
302    }
303
304    #[instrument(skip_all)]
305    async fn get_mint_quote(&self, quote_id: &str) -> Result<Option<MintQuote>, database::Error> {
306        let read_txn = self.db.begin_read().map_err(Into::<Error>::into)?;
307        let table = read_txn
308            .open_table(MINT_QUOTES_TABLE)
309            .map_err(Error::from)?;
310
311        if let Some(mint_info) = table.get(quote_id).map_err(Error::from)? {
312            return Ok(serde_json::from_str(mint_info.value()).map_err(Error::from)?);
313        }
314
315        Ok(None)
316    }
317
318    #[instrument(skip_all)]
319    async fn get_mint_quotes(&self) -> Result<Vec<MintQuote>, database::Error> {
320        let read_txn = self.db.begin_read().map_err(Into::<Error>::into)?;
321        let table = read_txn
322            .open_table(MINT_QUOTES_TABLE)
323            .map_err(Error::from)?;
324
325        Ok(table
326            .iter()
327            .map_err(Error::from)?
328            .flatten()
329            .flat_map(|(_id, quote)| serde_json::from_str(quote.value()))
330            .collect())
331    }
332
333    async fn get_unissued_mint_quotes(&self) -> Result<Vec<MintQuote>, database::Error> {
334        let read_txn = self.db.begin_read().map_err(Into::<Error>::into)?;
335        let table = read_txn
336            .open_table(MINT_QUOTES_TABLE)
337            .map_err(Error::from)?;
338
339        Ok(table
340            .iter()
341            .map_err(Error::from)?
342            .flatten()
343            .flat_map(|(_id, quote)| serde_json::from_str::<MintQuote>(quote.value()).ok())
344            .filter(|quote| {
345                quote.amount_issued == Amount::ZERO
346                    || quote.payment_method == PaymentMethod::Known(KnownMethod::Bolt12)
347            })
348            .collect())
349    }
350
351    #[instrument(skip_all)]
352    async fn get_melt_quote(
353        &self,
354        quote_id: &str,
355    ) -> Result<Option<wallet::MeltQuote>, database::Error> {
356        let read_txn = self.db.begin_read().map_err(Error::from)?;
357        let table = read_txn
358            .open_table(MELT_QUOTES_TABLE)
359            .map_err(Error::from)?;
360
361        if let Some(mint_info) = table.get(quote_id).map_err(Error::from)? {
362            return Ok(serde_json::from_str(mint_info.value()).map_err(Error::from)?);
363        }
364
365        Ok(None)
366    }
367
368    #[instrument(skip_all)]
369    async fn get_melt_quotes(&self) -> Result<Vec<wallet::MeltQuote>, database::Error> {
370        let read_txn = self.db.begin_read().map_err(Error::from)?;
371        let table = read_txn
372            .open_table(MELT_QUOTES_TABLE)
373            .map_err(Error::from)?;
374
375        Ok(table
376            .iter()
377            .map_err(Error::from)?
378            .flatten()
379            .flat_map(|(_id, quote)| serde_json::from_str(quote.value()))
380            .collect())
381    }
382
383    #[instrument(skip(self), fields(keyset_id = %keyset_id))]
384    async fn get_keys(&self, keyset_id: &Id) -> Result<Option<Keys>, database::Error> {
385        let read_txn = self.db.begin_read().map_err(Error::from)?;
386        let table = read_txn.open_table(MINT_KEYS_TABLE).map_err(Error::from)?;
387
388        if let Some(mint_info) = table
389            .get(keyset_id.to_string().as_str())
390            .map_err(Error::from)?
391        {
392            return Ok(serde_json::from_str(mint_info.value()).map_err(Error::from)?);
393        }
394
395        Ok(None)
396    }
397
398    #[instrument(skip_all)]
399    async fn get_proofs(
400        &self,
401        mint_url: Option<MintUrl>,
402        unit: Option<CurrencyUnit>,
403        state: Option<Vec<State>>,
404        spending_conditions: Option<Vec<SpendingConditions>>,
405    ) -> Result<Vec<ProofInfo>, database::Error> {
406        let read_txn = self.db.begin_read().map_err(Error::from)?;
407
408        let table = read_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
409
410        let proofs: Vec<ProofInfo> = table
411            .iter()
412            .map_err(Error::from)?
413            .flatten()
414            .filter_map(|(_k, v)| {
415                let mut proof = None;
416
417                if let Ok(proof_info) = serde_json::from_str::<ProofInfo>(v.value()) {
418                    if proof_info.matches_conditions(&mint_url, &unit, &state, &spending_conditions)
419                    {
420                        proof = Some(proof_info)
421                    }
422                }
423
424                proof
425            })
426            .collect();
427
428        Ok(proofs)
429    }
430
431    #[instrument(skip(self, ys))]
432    async fn get_proofs_by_ys(
433        &self,
434        ys: Vec<PublicKey>,
435    ) -> Result<Vec<ProofInfo>, database::Error> {
436        if ys.is_empty() {
437            return Ok(Vec::new());
438        }
439
440        let read_txn = self.db.begin_read().map_err(Error::from)?;
441        let table = read_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
442
443        let mut proofs = Vec::new();
444
445        for y in ys {
446            if let Some(proof) = table.get(y.to_bytes().as_slice()).map_err(Error::from)? {
447                let proof_info =
448                    serde_json::from_str::<ProofInfo>(proof.value()).map_err(Error::from)?;
449                proofs.push(proof_info);
450            }
451        }
452
453        Ok(proofs)
454    }
455
456    async fn get_balance(
457        &self,
458        mint_url: Option<MintUrl>,
459        unit: Option<CurrencyUnit>,
460        state: Option<Vec<State>>,
461    ) -> Result<u64, database::Error> {
462        // For redb, we still need to fetch all proofs and sum them
463        // since redb doesn't have SQL aggregation
464        let proofs = self.get_proofs(mint_url, unit, state, None).await?;
465        Ok(proofs.iter().map(|p| u64::from(p.proof.amount)).sum())
466    }
467
468    #[instrument(skip(self))]
469    async fn get_transaction(
470        &self,
471        transaction_id: TransactionId,
472    ) -> Result<Option<Transaction>, database::Error> {
473        let read_txn = self.db.begin_read().map_err(Error::from)?;
474        let table = read_txn
475            .open_table(TRANSACTIONS_TABLE)
476            .map_err(Error::from)?;
477
478        if let Some(transaction) = table.get(transaction_id.as_slice()).map_err(Error::from)? {
479            return Ok(serde_json::from_str(transaction.value()).map_err(Error::from)?);
480        }
481
482        Ok(None)
483    }
484
485    #[instrument(skip(self))]
486    async fn list_transactions(
487        &self,
488        mint_url: Option<MintUrl>,
489        direction: Option<TransactionDirection>,
490        unit: Option<CurrencyUnit>,
491    ) -> Result<Vec<Transaction>, database::Error> {
492        let read_txn = self.db.begin_read().map_err(Error::from)?;
493
494        let table = read_txn
495            .open_table(TRANSACTIONS_TABLE)
496            .map_err(Error::from)?;
497
498        let transactions: Vec<Transaction> = table
499            .iter()
500            .map_err(Error::from)?
501            .flatten()
502            .filter_map(|(_k, v)| {
503                let mut transaction = None;
504
505                if let Ok(tx) = serde_json::from_str::<Transaction>(v.value()) {
506                    if tx.matches_conditions(&mint_url, &direction, &unit) {
507                        transaction = Some(tx)
508                    }
509                }
510
511                transaction
512            })
513            .collect();
514
515        Ok(transactions)
516    }
517
518    #[instrument(skip(self, added, removed_ys))]
519    async fn update_proofs(
520        &self,
521        added: Vec<ProofInfo>,
522        removed_ys: Vec<PublicKey>,
523    ) -> Result<(), database::Error> {
524        let write_txn = self.db.begin_write().map_err(Error::from)?;
525        {
526            let mut table = write_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
527
528            for mut proof_info in added {
529                if proof_info.derivation_index.is_none() {
530                    proof_info.derivation_index = match table
531                        .get(proof_info.y.to_bytes().as_slice())
532                        .map_err(Error::from)?
533                    {
534                        Some(stored) => {
535                            serde_json::from_str::<ProofInfo>(stored.value())
536                                .map_err(Error::from)?
537                                .derivation_index
538                        }
539                        None => None,
540                    };
541                }
542
543                table
544                    .insert(
545                        proof_info.y.to_bytes().as_slice(),
546                        serde_json::to_string(&proof_info)
547                            .map_err(Error::from)?
548                            .as_str(),
549                    )
550                    .map_err(Error::from)?;
551            }
552
553            for y in removed_ys.iter() {
554                table.remove(y.to_bytes().as_slice()).map_err(Error::from)?;
555            }
556        }
557        write_txn.commit().map_err(Error::from)?;
558        Ok(())
559    }
560
561    async fn update_proofs_state(
562        &self,
563        ys: Vec<PublicKey>,
564        state: State,
565    ) -> Result<(), database::Error> {
566        let write_txn = self.db.begin_write().map_err(Error::from)?;
567        {
568            let mut table = write_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
569
570            for y in ys {
571                let y_slice = y.to_bytes();
572                let proof = table
573                    .get(y_slice.as_slice())
574                    .map_err(Error::from)?
575                    .ok_or(Error::UnknownY)?;
576
577                let mut proof_info =
578                    serde_json::from_str::<ProofInfo>(proof.value()).map_err(Error::from)?;
579                drop(proof);
580
581                proof_info.state = state;
582
583                table
584                    .insert(
585                        y_slice.as_slice(),
586                        serde_json::to_string(&proof_info)
587                            .map_err(Error::from)?
588                            .as_str(),
589                    )
590                    .map_err(Error::from)?;
591            }
592        }
593        write_txn.commit().map_err(Error::from)?;
594        Ok(())
595    }
596
597    #[instrument(
598        skip(self, transaction),
599        fields(
600            direction = %transaction.direction,
601            amount = %transaction.amount,
602            unit = %transaction.unit,
603            quote_id = ?transaction.quote_id,
604            saga_id = ?transaction.saga_id,
605        )
606    )]
607    async fn add_transaction(&self, transaction: Transaction) -> Result<(), database::Error> {
608        let id = transaction.id();
609        let write_txn = self.db.begin_write().map_err(Error::from)?;
610        {
611            let mut table = write_txn
612                .open_table(TRANSACTIONS_TABLE)
613                .map_err(Error::from)?;
614            table
615                .insert(
616                    id.as_slice(),
617                    serde_json::to_string(&transaction)
618                        .map_err(Error::from)?
619                        .as_str(),
620                )
621                .map_err(Error::from)?;
622        }
623        write_txn.commit().map_err(Error::from)?;
624        Ok(())
625    }
626
627    #[instrument(skip(self))]
628    async fn update_mint_url(
629        &self,
630        old_mint_url: MintUrl,
631        new_mint_url: MintUrl,
632    ) -> Result<(), database::Error> {
633        let write_txn = self.db.begin_write().map_err(Error::from)?;
634
635        // Update proofs table
636        {
637            let read_table = write_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
638            let proofs: Vec<ProofInfo> = read_table
639                .iter()
640                .map_err(Error::from)?
641                .flatten()
642                .filter_map(|(_k, v)| {
643                    let proof_info = serde_json::from_str::<ProofInfo>(v.value()).ok()?;
644                    if proof_info.mint_url == old_mint_url {
645                        Some(proof_info)
646                    } else {
647                        None
648                    }
649                })
650                .collect();
651            drop(read_table);
652
653            if !proofs.is_empty() {
654                let mut write_table = write_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
655                for mut proof_info in proofs {
656                    proof_info.mint_url = new_mint_url.clone();
657                    write_table
658                        .insert(
659                            proof_info.y.to_bytes().as_slice(),
660                            serde_json::to_string(&proof_info)
661                                .map_err(Error::from)?
662                                .as_str(),
663                        )
664                        .map_err(Error::from)?;
665                }
666            }
667        }
668
669        // Update mint quotes
670        {
671            let mut table = write_txn
672                .open_table(MINT_QUOTES_TABLE)
673                .map_err(Error::from)?;
674
675            let unix_time = unix_time();
676
677            let quotes: Vec<MintQuote> = table
678                .iter()
679                .map_err(Error::from)?
680                .flatten()
681                .filter_map(|(_, quote)| {
682                    let mut q: MintQuote = serde_json::from_str(quote.value()).ok()?;
683                    if q.mint_url == old_mint_url && q.expiry >= unix_time {
684                        q.mint_url = new_mint_url.clone();
685                        Some(q)
686                    } else {
687                        None
688                    }
689                })
690                .collect();
691
692            for quote in quotes {
693                table
694                    .insert(
695                        quote.id.as_str(),
696                        serde_json::to_string(&quote).map_err(Error::from)?.as_str(),
697                    )
698                    .map_err(Error::from)?;
699            }
700        }
701
702        write_txn.commit().map_err(Error::from)?;
703        Ok(())
704    }
705
706    #[instrument(skip(self), fields(keyset_id = %keyset_id))]
707    async fn increment_keyset_counter(
708        &self,
709        keyset_id: &Id,
710        count: u32,
711    ) -> Result<u32, database::Error> {
712        let write_txn = self.db.begin_write().map_err(Error::from)?;
713        let new_counter = {
714            let mut table = write_txn.open_table(KEYSET_COUNTER).map_err(Error::from)?;
715            let current_counter = table
716                .get(keyset_id.to_string().as_str())
717                .map_err(Error::from)?
718                .map(|x| x.value())
719                .unwrap_or_default();
720
721            let new_counter = current_counter
722                .checked_add(count)
723                .ok_or(database::Error::AmountOverflow)?;
724
725            table
726                .insert(keyset_id.to_string().as_str(), new_counter)
727                .map_err(Error::from)?;
728
729            new_counter
730        };
731        write_txn.commit().map_err(Error::from)?;
732        Ok(new_counter)
733    }
734
735    #[instrument(skip(self))]
736    async fn increment_derivation_counter(
737        &self,
738        namespace: &str,
739        count: u32,
740    ) -> Result<u32, database::Error> {
741        let write_txn = self.db.begin_write().map_err(Error::from)?;
742        let new_counter = {
743            let mut table = write_txn
744                .open_table(DERIVATION_COUNTER)
745                .map_err(Error::from)?;
746            let current_counter = table
747                .get(namespace)
748                .map_err(Error::from)?
749                .map(|value| value.value())
750                .unwrap_or_default();
751            let new_counter = current_counter
752                .checked_add(count)
753                .ok_or(database::Error::AmountOverflow)?;
754
755            table.insert(namespace, new_counter).map_err(Error::from)?;
756            new_counter
757        };
758        write_txn.commit().map_err(Error::from)?;
759
760        Ok(new_counter)
761    }
762
763    #[instrument(skip(self))]
764    async fn add_mint(
765        &self,
766        mint_url: MintUrl,
767        mint_info: Option<MintInfo>,
768    ) -> Result<(), database::Error> {
769        let write_txn = self.db.begin_write().map_err(Error::from)?;
770        {
771            let mut table = write_txn.open_table(MINTS_TABLE).map_err(Error::from)?;
772            table
773                .insert(
774                    mint_url.to_string().as_str(),
775                    serde_json::to_string(&mint_info)
776                        .map_err(Error::from)?
777                        .as_str(),
778                )
779                .map_err(Error::from)?;
780        }
781        write_txn.commit().map_err(Error::from)?;
782        Ok(())
783    }
784
785    #[instrument(skip(self))]
786    async fn remove_mint(&self, mint_url: MintUrl) -> Result<(), database::Error> {
787        let write_txn = self.db.begin_write().map_err(Error::from)?;
788        {
789            let mut table = write_txn.open_table(MINTS_TABLE).map_err(Error::from)?;
790            table
791                .remove(mint_url.to_string().as_str())
792                .map_err(Error::from)?;
793        }
794        write_txn.commit().map_err(Error::from)?;
795        Ok(())
796    }
797
798    #[instrument(skip(self))]
799    async fn add_mint_keysets(
800        &self,
801        mint_url: MintUrl,
802        keysets: Vec<KeySetInfo>,
803    ) -> Result<(), database::Error> {
804        let write_txn = self.db.begin_write().map_err(Error::from)?;
805        {
806            let mut table = write_txn
807                .open_multimap_table(MINT_KEYSETS_TABLE)
808                .map_err(Error::from)?;
809            let mut keysets_table = write_txn.open_table(KEYSETS_TABLE).map_err(Error::from)?;
810            let mut u32_table = write_txn
811                .open_table(KEYSET_U32_MAPPING)
812                .map_err(Error::from)?;
813
814            let mut existing_u32 = false;
815
816            for keyset in keysets {
817                // Check if keyset already exists
818                let existing_keyset = {
819                    let existing_keyset = keysets_table
820                        .get(keyset.id.to_bytes().as_slice())
821                        .map_err(Error::from)?;
822
823                    existing_keyset.map(|r| r.value().to_string())
824                };
825
826                let existing = u32_table
827                    .insert(u32::from(keyset.id), keyset.id.to_string().as_str())
828                    .map_err(Error::from)?;
829
830                match existing {
831                    None => existing_u32 = false,
832                    Some(id) => {
833                        let id = Id::from_str(id.value())?;
834
835                        if id == keyset.id {
836                            existing_u32 = false;
837                        } else {
838                            existing_u32 = true;
839                            break;
840                        }
841                    }
842                }
843
844                let keyset = if let Some(existing_keyset) = existing_keyset {
845                    let mut existing_keyset: KeySetInfo = serde_json::from_str(&existing_keyset)?;
846
847                    existing_keyset.active = keyset.active;
848                    existing_keyset.input_fee_ppk = keyset.input_fee_ppk;
849
850                    existing_keyset
851                } else {
852                    table
853                        .insert(
854                            mint_url.to_string().as_str(),
855                            keyset.id.to_bytes().as_slice(),
856                        )
857                        .map_err(Error::from)?;
858
859                    keyset
860                };
861
862                keysets_table
863                    .insert(
864                        keyset.id.to_bytes().as_slice(),
865                        serde_json::to_string(&keyset)
866                            .map_err(Error::from)?
867                            .as_str(),
868                    )
869                    .map_err(Error::from)?;
870            }
871
872            if existing_u32 {
873                tracing::warn!("Keyset already exists for keyset id");
874                return Err(database::Error::Duplicate);
875            }
876        }
877        write_txn.commit().map_err(Error::from)?;
878        Ok(())
879    }
880
881    #[instrument(skip_all)]
882    async fn add_mint_quote(&self, quote: MintQuote) -> Result<(), database::Error> {
883        let write_txn = self.db.begin_write().map_err(Error::from)?;
884        {
885            let mut table = write_txn
886                .open_table(MINT_QUOTES_TABLE)
887                .map_err(Error::from)?;
888
889            // Check for existing quote and version match
890            let existing_quote_json = table
891                .get(quote.id.as_str())
892                .map_err(Error::from)?
893                .map(|v| v.value().to_string());
894
895            let mut quote_to_save = quote.clone();
896
897            if let Some(json) = existing_quote_json {
898                let existing_quote: MintQuote = serde_json::from_str(&json).map_err(Error::from)?;
899
900                if existing_quote.version != quote.version {
901                    return Err(database::Error::ConcurrentUpdate);
902                }
903
904                // Increment version for update
905                quote_to_save.version = quote.version.wrapping_add(1);
906            }
907
908            table
909                .insert(
910                    quote_to_save.id.as_str(),
911                    serde_json::to_string(&quote_to_save)
912                        .map_err(Error::from)?
913                        .as_str(),
914                )
915                .map_err(Error::from)?;
916        }
917        write_txn.commit().map_err(Error::from)?;
918        Ok(())
919    }
920
921    #[instrument(skip_all)]
922    async fn remove_mint_quote(&self, quote_id: &str) -> Result<(), database::Error> {
923        let write_txn = self.db.begin_write().map_err(Error::from)?;
924        {
925            let mut table = write_txn
926                .open_table(MINT_QUOTES_TABLE)
927                .map_err(Error::from)?;
928            table.remove(quote_id).map_err(Error::from)?;
929        }
930        write_txn.commit().map_err(Error::from)?;
931        Ok(())
932    }
933
934    #[instrument(skip_all)]
935    async fn add_melt_quote(&self, quote: wallet::MeltQuote) -> Result<(), database::Error> {
936        let write_txn = self.db.begin_write().map_err(Error::from)?;
937        {
938            let mut table = write_txn
939                .open_table(MELT_QUOTES_TABLE)
940                .map_err(Error::from)?;
941
942            // Check for existing quote and version match
943            let existing_quote_json = table
944                .get(quote.id.as_str())
945                .map_err(Error::from)?
946                .map(|v| v.value().to_string());
947
948            let mut quote_to_save = quote.clone();
949
950            if let Some(json) = existing_quote_json {
951                let existing_quote: wallet::MeltQuote =
952                    serde_json::from_str(&json).map_err(Error::from)?;
953
954                if existing_quote.version != quote.version {
955                    return Err(database::Error::ConcurrentUpdate);
956                }
957
958                // Increment version for update
959                quote_to_save.version = quote.version.wrapping_add(1);
960            }
961
962            table
963                .insert(
964                    quote_to_save.id.as_str(),
965                    serde_json::to_string(&quote_to_save)
966                        .map_err(Error::from)?
967                        .as_str(),
968                )
969                .map_err(Error::from)?;
970        }
971        write_txn.commit().map_err(Error::from)?;
972        Ok(())
973    }
974
975    #[instrument(skip_all)]
976    async fn remove_melt_quote(&self, quote_id: &str) -> Result<(), database::Error> {
977        let write_txn = self.db.begin_write().map_err(Error::from)?;
978        {
979            let mut table = write_txn
980                .open_table(MELT_QUOTES_TABLE)
981                .map_err(Error::from)?;
982            table.remove(quote_id).map_err(Error::from)?;
983        }
984        write_txn.commit().map_err(Error::from)?;
985        Ok(())
986    }
987
988    #[instrument(skip_all)]
989    async fn add_keys(&self, keyset: KeySet) -> Result<(), database::Error> {
990        let write_txn = self.db.begin_write().map_err(Error::from)?;
991
992        keyset.verify_id()?;
993
994        {
995            let mut table = write_txn.open_table(MINT_KEYS_TABLE).map_err(Error::from)?;
996
997            let existing_keys = table
998                .insert(
999                    keyset.id.to_string().as_str(),
1000                    serde_json::to_string(&keyset.keys)
1001                        .map_err(Error::from)?
1002                        .as_str(),
1003                )
1004                .map_err(Error::from)?
1005                .is_some();
1006
1007            let mut table = write_txn
1008                .open_table(KEYSET_U32_MAPPING)
1009                .map_err(Error::from)?;
1010
1011            let existing = table
1012                .insert(u32::from(keyset.id), keyset.id.to_string().as_str())
1013                .map_err(Error::from)?;
1014
1015            let existing_u32 = match existing {
1016                None => false,
1017                Some(id) => {
1018                    let id = Id::from_str(id.value())?;
1019                    id != keyset.id
1020                }
1021            };
1022
1023            if existing_keys || existing_u32 {
1024                tracing::warn!("Keys already exist for keyset id");
1025                return Err(database::Error::Duplicate);
1026            }
1027        }
1028        write_txn.commit().map_err(Error::from)?;
1029        Ok(())
1030    }
1031
1032    #[instrument(skip(self), fields(keyset_id = %keyset_id))]
1033    async fn remove_keys(&self, keyset_id: &Id) -> Result<(), database::Error> {
1034        let write_txn = self.db.begin_write().map_err(Error::from)?;
1035        {
1036            let mut table = write_txn.open_table(MINT_KEYS_TABLE).map_err(Error::from)?;
1037
1038            table
1039                .remove(keyset_id.to_string().as_str())
1040                .map_err(Error::from)?;
1041        }
1042        write_txn.commit().map_err(Error::from)?;
1043        Ok(())
1044    }
1045
1046    #[instrument(skip(self))]
1047    async fn remove_transaction(
1048        &self,
1049        transaction_id: TransactionId,
1050    ) -> Result<(), database::Error> {
1051        let write_txn = self.db.begin_write().map_err(Error::from)?;
1052        {
1053            let mut table = write_txn
1054                .open_table(TRANSACTIONS_TABLE)
1055                .map_err(Error::from)?;
1056            table
1057                .remove(transaction_id.as_slice())
1058                .map_err(Error::from)?;
1059        }
1060        write_txn.commit().map_err(Error::from)?;
1061        Ok(())
1062    }
1063
1064    #[instrument(skip(self))]
1065    async fn add_saga(&self, saga: wallet::WalletSaga) -> Result<(), database::Error> {
1066        let saga_json = serde_json::to_string(&saga).map_err(Error::from)?;
1067        let id_str = saga.id.to_string();
1068
1069        let write_txn = self.db.begin_write().map_err(Error::from)?;
1070        {
1071            let mut table = write_txn.open_table(SAGAS_TABLE).map_err(Error::from)?;
1072            table
1073                .insert(id_str.as_str(), saga_json.as_str())
1074                .map_err(Error::from)?;
1075        }
1076        write_txn.commit().map_err(Error::from)?;
1077        Ok(())
1078    }
1079
1080    #[instrument(skip(self))]
1081    async fn get_saga(
1082        &self,
1083        id: &uuid::Uuid,
1084    ) -> Result<Option<wallet::WalletSaga>, database::Error> {
1085        let read_txn = self.db.begin_read().map_err(Error::from)?;
1086        let table = read_txn.open_table(SAGAS_TABLE).map_err(Error::from)?;
1087        let id_str = id.to_string();
1088
1089        let result = table
1090            .get(id_str.as_str())
1091            .map_err(Error::from)?
1092            .map(|saga| serde_json::from_str(saga.value()).map_err(Error::from))
1093            .transpose()?;
1094
1095        Ok(result)
1096    }
1097
1098    #[instrument(skip(self))]
1099    async fn update_saga(&self, saga: wallet::WalletSaga) -> Result<bool, database::Error> {
1100        let id_str = saga.id.to_string();
1101
1102        // The saga.version has already been incremented by the caller, so we check
1103        // for (saga.version - 1) as the expected version in the database.
1104        let expected_version = saga.version.saturating_sub(1);
1105
1106        let write_txn = self.db.begin_write().map_err(Error::from)?;
1107        let updated = {
1108            let mut table = write_txn.open_table(SAGAS_TABLE).map_err(Error::from)?;
1109
1110            // Read existing saga to check version (optimistic locking)
1111            let existing_saga_json = table
1112                .get(id_str.as_str())
1113                .map_err(Error::from)?
1114                .map(|v| v.value().to_string());
1115
1116            match existing_saga_json {
1117                Some(json) => {
1118                    let existing_saga: wallet::WalletSaga =
1119                        serde_json::from_str(&json).map_err(Error::from)?;
1120
1121                    // Check if version matches expected version
1122                    if existing_saga.version != expected_version {
1123                        // Version mismatch - another instance modified it
1124                        false
1125                    } else {
1126                        // Version matches - safe to update
1127                        let saga_json = serde_json::to_string(&saga).map_err(Error::from)?;
1128                        table
1129                            .insert(id_str.as_str(), saga_json.as_str())
1130                            .map_err(Error::from)?;
1131                        true
1132                    }
1133                }
1134                None => {
1135                    // Saga doesn't exist - can't update
1136                    false
1137                }
1138            }
1139        };
1140        write_txn.commit().map_err(Error::from)?;
1141        Ok(updated)
1142    }
1143
1144    #[instrument(skip(self))]
1145    async fn delete_saga(&self, id: &uuid::Uuid) -> Result<(), database::Error> {
1146        let write_txn = self.db.begin_write().map_err(Error::from)?;
1147        let id_str = id.to_string();
1148        {
1149            let mut table = write_txn.open_table(SAGAS_TABLE).map_err(Error::from)?;
1150            table.remove(id_str.as_str()).map_err(Error::from)?;
1151        }
1152        write_txn.commit().map_err(Error::from)?;
1153        Ok(())
1154    }
1155
1156    #[instrument(skip(self))]
1157    async fn get_incomplete_sagas(&self) -> Result<Vec<wallet::WalletSaga>, database::Error> {
1158        let read_txn = self.db.begin_read().map_err(Error::from)?;
1159        let table = read_txn.open_table(SAGAS_TABLE).map_err(Error::from)?;
1160
1161        let mut sagas: Vec<wallet::WalletSaga> = table
1162            .iter()
1163            .map_err(Error::from)?
1164            .flatten()
1165            .filter_map(|(_, saga_json)| {
1166                serde_json::from_str::<wallet::WalletSaga>(saga_json.value()).ok()
1167            })
1168            .collect();
1169
1170        // Sort by created_at ascending (oldest first)
1171        sagas.sort_by_key(|saga| saga.created_at);
1172
1173        Ok(sagas)
1174    }
1175
1176    #[instrument(skip(self))]
1177    async fn reserve_proofs(
1178        &self,
1179        ys: Vec<PublicKey>,
1180        operation_id: &uuid::Uuid,
1181    ) -> Result<(), database::Error> {
1182        let write_txn = self.db.begin_write().map_err(Error::from)?;
1183
1184        {
1185            let mut table = write_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
1186
1187            for y in ys {
1188                let y_bytes = y.to_bytes();
1189
1190                // Read the proof and convert to string immediately
1191                let proof_json_str = {
1192                    let proof_json_opt = table.get(y_bytes.as_slice()).map_err(Error::from)?;
1193                    proof_json_opt.map(|proof_json| proof_json.value().to_string())
1194                };
1195
1196                let Some(proof_json_str) = proof_json_str else {
1197                    return Err(database::Error::ProofNotUnspent);
1198                };
1199
1200                let mut proof: ProofInfo =
1201                    serde_json::from_str(&proof_json_str).map_err(Error::from)?;
1202
1203                if proof.state != State::Unspent {
1204                    return Err(database::Error::ProofNotUnspent);
1205                }
1206
1207                proof.state = State::Reserved;
1208                proof.used_by_operation = Some(*operation_id);
1209
1210                let updated_json = serde_json::to_string(&proof).map_err(Error::from)?;
1211                table
1212                    .insert(y_bytes.as_slice(), updated_json.as_str())
1213                    .map_err(Error::from)?;
1214            }
1215        }
1216
1217        write_txn.commit().map_err(Error::from)?;
1218        Ok(())
1219    }
1220
1221    #[instrument(skip(self))]
1222    async fn release_proofs(&self, operation_id: &uuid::Uuid) -> Result<(), database::Error> {
1223        let write_txn = self.db.begin_write().map_err(Error::from)?;
1224
1225        {
1226            let mut table = write_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
1227
1228            // Collect all proofs first to avoid borrowing issues
1229            let all_proofs: Vec<(Vec<u8>, ProofInfo)> = table
1230                .iter()
1231                .map_err(Error::from)?
1232                .flatten()
1233                .filter_map(|(y, proof_json)| {
1234                    let proof: ProofInfo = serde_json::from_str(proof_json.value()).ok()?;
1235                    Some((y.value().to_vec(), proof))
1236                })
1237                .collect();
1238
1239            // Now update proofs that match the operation_id
1240            for (y_bytes, mut proof) in all_proofs {
1241                if proof.used_by_operation == Some(*operation_id)
1242                    && matches!(proof.state, State::Reserved | State::Pending)
1243                {
1244                    proof.state = State::Unspent;
1245                    proof.used_by_operation = None;
1246
1247                    let updated_json = serde_json::to_string(&proof).map_err(Error::from)?;
1248                    table
1249                        .insert(y_bytes.as_slice(), updated_json.as_str())
1250                        .map_err(Error::from)?;
1251                }
1252            }
1253        }
1254
1255        write_txn.commit().map_err(Error::from)?;
1256        Ok(())
1257    }
1258
1259    #[instrument(skip(self))]
1260    async fn get_reserved_proofs(
1261        &self,
1262        operation_id: &uuid::Uuid,
1263    ) -> Result<Vec<ProofInfo>, database::Error> {
1264        let read_txn = self.db.begin_read().map_err(Error::from)?;
1265        let table = read_txn.open_table(PROOFS_TABLE).map_err(Error::from)?;
1266
1267        let proofs: Vec<ProofInfo> = table
1268            .iter()
1269            .map_err(Error::from)?
1270            .flatten()
1271            .filter_map(|(_, proof_json)| {
1272                serde_json::from_str::<ProofInfo>(proof_json.value()).ok()
1273            })
1274            .filter(|proof| proof.used_by_operation == Some(*operation_id))
1275            .collect();
1276
1277        Ok(proofs)
1278    }
1279
1280    #[instrument(skip(self))]
1281    async fn reserve_melt_quote(
1282        &self,
1283        quote_id: &str,
1284        operation_id: &uuid::Uuid,
1285    ) -> Result<(), database::Error> {
1286        let write_txn = self.db.begin_write().map_err(Error::from)?;
1287        let operation_id_str = operation_id.to_string();
1288
1289        {
1290            let mut table = write_txn
1291                .open_table(MELT_QUOTES_TABLE)
1292                .map_err(Error::from)?;
1293
1294            // Read existing quote
1295            let quote_json = table
1296                .get(quote_id)
1297                .map_err(Error::from)?
1298                .map(|v| v.value().to_string());
1299
1300            match quote_json {
1301                Some(json) => {
1302                    let mut quote: wallet::MeltQuote =
1303                        serde_json::from_str(&json).map_err(Error::from)?;
1304
1305                    // Check if already reserved by another operation
1306                    if quote.used_by_operation.is_some() {
1307                        return Err(database::Error::QuoteAlreadyInUse);
1308                    }
1309
1310                    // Reserve the quote
1311                    quote.used_by_operation = Some(operation_id_str);
1312                    let updated_json = serde_json::to_string(&quote).map_err(Error::from)?;
1313                    table
1314                        .insert(quote_id, updated_json.as_str())
1315                        .map_err(Error::from)?;
1316                }
1317                None => {
1318                    return Err(database::Error::UnknownQuote);
1319                }
1320            }
1321        }
1322
1323        write_txn.commit().map_err(Error::from)?;
1324        Ok(())
1325    }
1326
1327    #[instrument(skip(self))]
1328    async fn release_melt_quote(&self, operation_id: &uuid::Uuid) -> Result<(), database::Error> {
1329        let write_txn = self.db.begin_write().map_err(Error::from)?;
1330        let operation_id_str = operation_id.to_string();
1331
1332        {
1333            let mut table = write_txn
1334                .open_table(MELT_QUOTES_TABLE)
1335                .map_err(Error::from)?;
1336
1337            // Collect all quotes first to avoid borrowing issues
1338            let all_quotes: Vec<(String, wallet::MeltQuote)> = table
1339                .iter()
1340                .map_err(Error::from)?
1341                .flatten()
1342                .filter_map(|(id, quote_json)| {
1343                    let quote: wallet::MeltQuote = serde_json::from_str(quote_json.value()).ok()?;
1344                    Some((id.value().to_string(), quote))
1345                })
1346                .collect();
1347
1348            // Update quotes that match the operation_id
1349            for (quote_id, mut quote) in all_quotes {
1350                if quote.used_by_operation.as_deref() == Some(&operation_id_str) {
1351                    quote.used_by_operation = None;
1352                    let updated_json = serde_json::to_string(&quote).map_err(Error::from)?;
1353                    table
1354                        .insert(quote_id.as_str(), updated_json.as_str())
1355                        .map_err(Error::from)?;
1356                }
1357            }
1358        }
1359
1360        write_txn.commit().map_err(Error::from)?;
1361        Ok(())
1362    }
1363
1364    #[instrument(skip(self))]
1365    async fn reserve_mint_quote(
1366        &self,
1367        quote_id: &str,
1368        operation_id: &uuid::Uuid,
1369    ) -> Result<(), database::Error> {
1370        let write_txn = self.db.begin_write().map_err(Error::from)?;
1371        let operation_id_str = operation_id.to_string();
1372
1373        {
1374            let mut table = write_txn
1375                .open_table(MINT_QUOTES_TABLE)
1376                .map_err(Error::from)?;
1377
1378            // Read existing quote
1379            let quote_json = table
1380                .get(quote_id)
1381                .map_err(Error::from)?
1382                .map(|v| v.value().to_string());
1383
1384            match quote_json {
1385                Some(json) => {
1386                    let mut quote: MintQuote = serde_json::from_str(&json).map_err(Error::from)?;
1387
1388                    // Check if already reserved by another operation
1389                    if quote.used_by_operation.is_some() {
1390                        return Err(database::Error::QuoteAlreadyInUse);
1391                    }
1392
1393                    // Reserve the quote
1394                    quote.used_by_operation = Some(operation_id_str);
1395                    let updated_json = serde_json::to_string(&quote).map_err(Error::from)?;
1396                    table
1397                        .insert(quote_id, updated_json.as_str())
1398                        .map_err(Error::from)?;
1399                }
1400                None => {
1401                    return Err(database::Error::UnknownQuote);
1402                }
1403            }
1404        }
1405
1406        write_txn.commit().map_err(Error::from)?;
1407        Ok(())
1408    }
1409
1410    #[instrument(skip(self))]
1411    async fn release_mint_quote(&self, operation_id: &uuid::Uuid) -> Result<(), database::Error> {
1412        let write_txn = self.db.begin_write().map_err(Error::from)?;
1413        let operation_id_str = operation_id.to_string();
1414
1415        {
1416            let mut table = write_txn
1417                .open_table(MINT_QUOTES_TABLE)
1418                .map_err(Error::from)?;
1419
1420            // Collect all quotes first to avoid borrowing issues
1421            let all_quotes: Vec<(String, MintQuote)> = table
1422                .iter()
1423                .map_err(Error::from)?
1424                .flatten()
1425                .filter_map(|(id, quote_json)| {
1426                    let quote: MintQuote = serde_json::from_str(quote_json.value()).ok()?;
1427                    Some((id.value().to_string(), quote))
1428                })
1429                .collect();
1430
1431            // Update quotes that match the operation_id
1432            for (quote_id, mut quote) in all_quotes {
1433                if quote.used_by_operation.as_deref() == Some(&operation_id_str) {
1434                    quote.used_by_operation = None;
1435                    let updated_json = serde_json::to_string(&quote).map_err(Error::from)?;
1436                    table
1437                        .insert(quote_id.as_str(), updated_json.as_str())
1438                        .map_err(Error::from)?;
1439                }
1440            }
1441        }
1442
1443        write_txn.commit().map_err(Error::from)?;
1444        Ok(())
1445    }
1446
1447    #[instrument(skip(self, value))]
1448    async fn kv_write(
1449        &self,
1450        primary_namespace: &str,
1451        secondary_namespace: &str,
1452        key: &str,
1453        value: &[u8],
1454    ) -> Result<(), database::Error> {
1455        // Validate parameters according to KV store requirements
1456        validate_kvstore_params(primary_namespace, secondary_namespace, Some(key))?;
1457
1458        let write_txn = self.db.begin_write().map_err(Error::from)?;
1459        {
1460            let mut table = write_txn.open_table(KV_STORE_TABLE).map_err(Error::from)?;
1461            table
1462                .insert((primary_namespace, secondary_namespace, key), value)
1463                .map_err(Error::from)?;
1464        }
1465        write_txn.commit().map_err(Error::from)?;
1466
1467        Ok(())
1468    }
1469
1470    #[instrument(skip(self))]
1471    async fn kv_read(
1472        &self,
1473        primary_namespace: &str,
1474        secondary_namespace: &str,
1475        key: &str,
1476    ) -> Result<Option<Vec<u8>>, database::Error> {
1477        // Validate parameters according to KV store requirements
1478        validate_kvstore_params(primary_namespace, secondary_namespace, Some(key))?;
1479
1480        let read_txn = self.db.begin_read().map_err(Error::from)?;
1481        let table = read_txn.open_table(KV_STORE_TABLE).map_err(Error::from)?;
1482
1483        let result = table
1484            .get((primary_namespace, secondary_namespace, key))
1485            .map_err(Error::from)?
1486            .map(|v| v.value().to_vec());
1487
1488        Ok(result)
1489    }
1490
1491    #[instrument(skip(self))]
1492    async fn kv_list(
1493        &self,
1494        primary_namespace: &str,
1495        secondary_namespace: &str,
1496    ) -> Result<Vec<String>, database::Error> {
1497        // Validate parameters according to KV store requirements
1498        validate_kvstore_params(primary_namespace, secondary_namespace, None)?;
1499
1500        let read_txn = self.db.begin_read().map_err(Error::from)?;
1501        let table = read_txn.open_table(KV_STORE_TABLE).map_err(Error::from)?;
1502
1503        let start = (primary_namespace, secondary_namespace, "");
1504        let iter = table.range(start..).map_err(Error::from)?;
1505
1506        let mut keys = Vec::new();
1507
1508        for item in iter {
1509            let (key, _) = item.map_err(Error::from)?;
1510            let (p, s, k) = key.value();
1511            if p == primary_namespace && s == secondary_namespace {
1512                keys.push(k.to_string());
1513            } else {
1514                break;
1515            }
1516        }
1517
1518        Ok(keys)
1519    }
1520
1521    #[instrument(skip(self))]
1522    async fn kv_remove(
1523        &self,
1524        primary_namespace: &str,
1525        secondary_namespace: &str,
1526        key: &str,
1527    ) -> Result<(), database::Error> {
1528        // Validate parameters according to KV store requirements
1529        validate_kvstore_params(primary_namespace, secondary_namespace, Some(key))?;
1530
1531        let write_txn = self.db.begin_write().map_err(Error::from)?;
1532        {
1533            let mut table = write_txn.open_table(KV_STORE_TABLE).map_err(Error::from)?;
1534            table
1535                .remove((primary_namespace, secondary_namespace, key))
1536                .map_err(Error::from)?;
1537        }
1538        write_txn.commit().map_err(Error::from)?;
1539
1540        Ok(())
1541    }
1542
1543    #[instrument(skip(self))]
1544    async fn add_p2pk_key(
1545        &self,
1546        pubkey: &PublicKey,
1547        derivation_path: DerivationPath,
1548        derivation_index: u32,
1549    ) -> Result<(), database::Error> {
1550        let write_txn = self.db.begin_write().map_err(Error::from)?;
1551        {
1552            let mut table = write_txn
1553                .open_table(P2PK_SIGNING_KEYS_TABLE)
1554                .map_err(Error::from)?;
1555            table
1556                .insert(
1557                    pubkey.to_bytes().as_slice(),
1558                    serde_json::to_string(&wallet::P2PKSigningKey {
1559                        pubkey: *pubkey,
1560                        derivation_path,
1561                        derivation_index,
1562                        created_time: unix_time(),
1563                    })
1564                    .map_err(Error::from)?
1565                    .as_str(),
1566                )
1567                .map_err(Error::from)?;
1568        }
1569        write_txn.commit().map_err(Error::from)?;
1570        Ok(())
1571    }
1572
1573    #[instrument(skip(self))]
1574    async fn get_p2pk_key(
1575        &self,
1576        pubkey: &PublicKey,
1577    ) -> Result<Option<wallet::P2PKSigningKey>, database::Error> {
1578        let read_txn = self.db.begin_read().map_err(Error::from)?;
1579        let table = read_txn
1580            .open_table(P2PK_SIGNING_KEYS_TABLE)
1581            .map_err(Error::from)?;
1582
1583        if let Some(key) = table
1584            .get(pubkey.to_bytes().as_slice())
1585            .map_err(Error::from)?
1586        {
1587            return Ok(Some(
1588                serde_json::from_str(key.value()).map_err(Error::from)?,
1589            ));
1590        }
1591
1592        Ok(None)
1593    }
1594
1595    #[instrument(skip(self))]
1596    async fn list_p2pk_keys(&self) -> Result<Vec<wallet::P2PKSigningKey>, database::Error> {
1597        let read_txn = self.db.begin_read().map_err(Error::from)?;
1598        let table = read_txn
1599            .open_table(P2PK_SIGNING_KEYS_TABLE)
1600            .map_err(Error::from)?;
1601
1602        let keys: Vec<wallet::P2PKSigningKey> = table
1603            .iter()
1604            .map_err(Error::from)?
1605            .flatten()
1606            .filter_map(|(_k, v)| {
1607                if let Ok(key) = serde_json::from_str::<wallet::P2PKSigningKey>(v.value()) {
1608                    return Some(key);
1609                }
1610
1611                None
1612            })
1613            .collect();
1614
1615        Ok(keys)
1616    }
1617
1618    #[instrument(skip(self))]
1619    async fn latest_p2pk(&self) -> Result<Option<wallet::P2PKSigningKey>, database::Error> {
1620        let read_txn = self.db.begin_read().map_err(Error::from)?;
1621        let table = read_txn
1622            .open_table(P2PK_SIGNING_KEYS_TABLE)
1623            .map_err(Error::from)?;
1624
1625        let latest_key = table
1626            .iter()
1627            .map_err(Error::from)?
1628            .flatten()
1629            .filter_map(|(_k, v)| serde_json::from_str::<wallet::P2PKSigningKey>(v.value()).ok())
1630            .max_by_key(|key| key.derivation_index);
1631
1632        Ok(latest_key)
1633    }
1634}
1635
1636#[cfg(test)]
1637mod test {
1638    use std::path::PathBuf;
1639    use std::str::FromStr;
1640
1641    use cdk_common::database::{self, WalletDatabase};
1642    use cdk_common::{wallet_db_test, Id};
1643
1644    use super::WalletRedbDatabase;
1645
1646    async fn provide_db(test_id: String) -> WalletRedbDatabase {
1647        let path = PathBuf::from(format!("/tmp/cdk-test-{}.redb", test_id));
1648        WalletRedbDatabase::new(&path).expect("database")
1649    }
1650
1651    wallet_db_test!(provide_db);
1652
1653    #[tokio::test]
1654    async fn increment_keyset_counter_returns_error_on_overflow() {
1655        let db = provide_db(format!("counter-overflow-{}", uuid::Uuid::new_v4())).await;
1656        let keyset_id = Id::from_str("00916bbf7ef91a36").expect("valid keyset id");
1657
1658        let first = db
1659            .increment_keyset_counter(&keyset_id, u32::MAX)
1660            .await
1661            .expect("first increment should fit");
1662        assert_eq!(first, u32::MAX);
1663
1664        match db.increment_keyset_counter(&keyset_id, 1).await {
1665            Err(database::Error::AmountOverflow) => {}
1666            Ok(counter) => panic!("counter should not wrap, got {counter}"),
1667            Err(err) => panic!("expected amount overflow, got {err}"),
1668        }
1669    }
1670}
cdk_redb/wallet/mod.rs

cdk_redb/wallet/
mod.rs
