cdk_bdk/storage/
send.rs

1use std::str::FromStr;
2
3use uuid::Uuid;
4
5use super::{
6    outpoint_to_key, BdkStorage, FailedSendAttemptRecord, FinalizedSendIntentRecord, BDK_NAMESPACE,
7    FINALIZED_INTENT_NAMESPACE, FINALIZED_SEND_INTENT_QUOTE_ID_NAMESPACE, SEND_INTENT_NAMESPACE,
8    SEND_INTENT_QUOTE_ID_NAMESPACE, SEND_OUTPOINT_QUOTE_ID_BACKFILL_KEY,
9    SEND_OUTPOINT_QUOTE_ID_NAMESPACE, STORAGE_MIGRATION_NAMESPACE,
10};
11use crate::error::Error;
12use crate::send::batch_transaction::record::{SendBatchRecord, SendBatchState};
13use crate::send::payment_intent::record::{SendIntentRecord, SendIntentState};
14
15impl BdkStorage {
16    // ── Send Intent storage ──────────────────────────────────────────
17
18    /// Store a new send intent and quote-id index atomically.
19    pub async fn create_send_intent_if_absent(
20        &self,
21        intent: &SendIntentRecord,
22    ) -> Result<(), Error> {
23        let mut tx = self
24            .kv_store
25            .begin_transaction()
26            .await
27            .map_err(Error::from)?;
28
29        let active = tx
30            .kv_read(
31                BDK_NAMESPACE,
32                SEND_INTENT_QUOTE_ID_NAMESPACE,
33                &intent.quote_id,
34            )
35            .await
36            .map_err(Error::from)?;
37
38        if active.is_some() {
39            tx.rollback().await.map_err(Error::from)?;
40            return Err(Error::DuplicateQuoteId(intent.quote_id.clone()));
41        }
42
43        let finalized = tx
44            .kv_read(
45                BDK_NAMESPACE,
46                FINALIZED_SEND_INTENT_QUOTE_ID_NAMESPACE,
47                &intent.quote_id,
48            )
49            .await
50            .map_err(Error::from)?;
51
52        if finalized.is_some() {
53            tx.rollback().await.map_err(Error::from)?;
54            return Err(Error::DuplicateQuoteId(intent.quote_id.clone()));
55        }
56
57        let serialized = serde_json::to_vec(intent)?;
58        tx.kv_write(
59            BDK_NAMESPACE,
60            SEND_INTENT_NAMESPACE,
61            &intent.intent_id.to_string(),
62            &serialized,
63        )
64        .await
65        .map_err(Error::from)?;
66        tx.kv_write(
67            BDK_NAMESPACE,
68            SEND_INTENT_QUOTE_ID_NAMESPACE,
69            &intent.quote_id,
70            intent.intent_id.to_string().as_bytes(),
71        )
72        .await
73        .map_err(Error::from)?;
74        if let SendIntentState::AwaitingConfirmation { outpoint, .. } = &intent.state {
75            tx.kv_write(
76                BDK_NAMESPACE,
77                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
78                &outpoint_to_key(outpoint),
79                intent.quote_id.as_bytes(),
80            )
81            .await
82            .map_err(Error::from)?;
83        }
84        tx.commit().await.map_err(Error::from)?;
85        Ok(())
86    }
87
88    /// Store a new send intent, or re-queue an existing failed intent with
89    /// the same quote id.
90    pub async fn create_or_retry_failed_send_intent(
91        &self,
92        intent: &SendIntentRecord,
93    ) -> Result<SendIntentRecord, Error> {
94        let mut tx = self
95            .kv_store
96            .begin_transaction()
97            .await
98            .map_err(Error::from)?;
99
100        let finalized = tx
101            .kv_read(
102                BDK_NAMESPACE,
103                FINALIZED_SEND_INTENT_QUOTE_ID_NAMESPACE,
104                &intent.quote_id,
105            )
106            .await
107            .map_err(Error::from)?;
108
109        if finalized.is_some() {
110            tx.rollback().await.map_err(Error::from)?;
111            return Err(Error::DuplicateQuoteId(intent.quote_id.clone()));
112        }
113
114        let active = tx
115            .kv_read(
116                BDK_NAMESPACE,
117                SEND_INTENT_QUOTE_ID_NAMESPACE,
118                &intent.quote_id,
119            )
120            .await
121            .map_err(Error::from)?;
122
123        let record = if let Some(intent_id_bytes) = active {
124            let intent_id_str = std::str::from_utf8(&intent_id_bytes)
125                .map_err(|e| Error::Wallet(format!("Invalid quote-id index entry: {}", e)))?;
126            let intent_id = Uuid::from_str(intent_id_str)
127                .map_err(|e| Error::Wallet(format!("Invalid indexed intent id: {}", e)))?;
128            let intent_bytes = tx
129                .kv_read(BDK_NAMESPACE, SEND_INTENT_NAMESPACE, &intent_id.to_string())
130                .await
131                .map_err(Error::from)?
132                .ok_or(Error::SendIntentNotFound(intent_id))?;
133            let existing: SendIntentRecord = serde_json::from_slice(&intent_bytes)?;
134
135            if !matches!(existing.state, SendIntentState::Failed { .. }) {
136                tx.rollback().await.map_err(Error::from)?;
137                return Err(Error::DuplicateQuoteId(intent.quote_id.clone()));
138            }
139
140            SendIntentRecord {
141                intent_id,
142                quote_id: intent.quote_id.clone(),
143                address: intent.address.clone(),
144                amount_sat: intent.amount_sat,
145                max_fee_amount_sat: intent.max_fee_amount_sat,
146                tier: intent.tier,
147                metadata: intent.metadata.clone(),
148                state: intent.state.clone(),
149            }
150        } else {
151            tx.kv_write(
152                BDK_NAMESPACE,
153                SEND_INTENT_QUOTE_ID_NAMESPACE,
154                &intent.quote_id,
155                intent.intent_id.to_string().as_bytes(),
156            )
157            .await
158            .map_err(Error::from)?;
159            intent.clone()
160        };
161
162        let serialized = serde_json::to_vec(&record)?;
163        tx.kv_write(
164            BDK_NAMESPACE,
165            SEND_INTENT_NAMESPACE,
166            &record.intent_id.to_string(),
167            &serialized,
168        )
169        .await
170        .map_err(Error::from)?;
171        if let SendIntentState::AwaitingConfirmation { outpoint, .. } = &record.state {
172            tx.kv_write(
173                BDK_NAMESPACE,
174                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
175                &outpoint_to_key(outpoint),
176                record.quote_id.as_bytes(),
177            )
178            .await
179            .map_err(Error::from)?;
180        }
181        tx.commit().await.map_err(Error::from)?;
182        Ok(record)
183    }
184
185    /// Get a send intent by ID
186    pub async fn get_send_intent(
187        &self,
188        intent_id: &Uuid,
189    ) -> Result<Option<SendIntentRecord>, Error> {
190        self.get_record::<SendIntentRecord>(&intent_id.to_string())
191            .await
192    }
193
194    /// Update a send intent's state
195    pub async fn update_send_intent(
196        &self,
197        intent_id: &Uuid,
198        new_state: &SendIntentState,
199    ) -> Result<(), Error> {
200        let Some(mut intent) = self.get_send_intent(intent_id).await? else {
201            return Err(Error::SendIntentNotFound(*intent_id));
202        };
203        let previous_outpoint = match &intent.state {
204            SendIntentState::AwaitingConfirmation { outpoint, .. } => Some(outpoint.clone()),
205            _ => None,
206        };
207        let new_outpoint = match new_state {
208            SendIntentState::AwaitingConfirmation { outpoint, .. } => Some(outpoint.clone()),
209            _ => None,
210        };
211        intent.state = new_state.clone();
212
213        let serialized = serde_json::to_vec(&intent)?;
214        let mut tx = self
215            .kv_store
216            .begin_transaction()
217            .await
218            .map_err(Error::from)?;
219        tx.kv_write(
220            BDK_NAMESPACE,
221            SEND_INTENT_NAMESPACE,
222            &intent_id.to_string(),
223            &serialized,
224        )
225        .await
226        .map_err(Error::from)?;
227        if let Some(outpoint) =
228            previous_outpoint.filter(|outpoint| Some(outpoint) != new_outpoint.as_ref())
229        {
230            tx.kv_remove(
231                BDK_NAMESPACE,
232                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
233                &outpoint_to_key(&outpoint),
234            )
235            .await
236            .map_err(Error::from)?;
237        }
238        if let Some(outpoint) = new_outpoint {
239            tx.kv_write(
240                BDK_NAMESPACE,
241                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
242                &outpoint_to_key(&outpoint),
243                intent.quote_id.as_bytes(),
244            )
245            .await
246            .map_err(Error::from)?;
247        }
248        tx.commit().await.map_err(Error::from)
249    }
250
251    /// Delete a send intent
252    pub async fn delete_send_intent(&self, intent_id: &Uuid) -> Result<(), Error> {
253        let Some(intent) = self.get_send_intent(intent_id).await? else {
254            return Ok(());
255        };
256
257        let mut tx = self
258            .kv_store
259            .begin_transaction()
260            .await
261            .map_err(Error::from)?;
262        tx.kv_remove(BDK_NAMESPACE, SEND_INTENT_NAMESPACE, &intent_id.to_string())
263            .await
264            .map_err(Error::from)?;
265        tx.kv_remove(
266            BDK_NAMESPACE,
267            SEND_INTENT_QUOTE_ID_NAMESPACE,
268            &intent.quote_id,
269        )
270        .await
271        .map_err(Error::from)?;
272        if let SendIntentState::AwaitingConfirmation { outpoint, .. } = intent.state {
273            tx.kv_remove(
274                BDK_NAMESPACE,
275                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
276                &outpoint_to_key(&outpoint),
277            )
278            .await
279            .map_err(Error::from)?;
280        }
281        tx.commit().await.map_err(Error::from)?;
282        Ok(())
283    }
284
285    /// Get all send intents
286    pub async fn get_all_send_intents(&self) -> Result<Vec<SendIntentRecord>, Error> {
287        self.list_records::<SendIntentRecord>().await
288    }
289
290    /// Get all pending send intents (filtering by state)
291    pub async fn get_pending_send_intents(&self) -> Result<Vec<SendIntentRecord>, Error> {
292        let all = self.get_all_send_intents().await?;
293        Ok(all
294            .into_iter()
295            .filter(|i| matches!(i.state, SendIntentState::Pending { .. }))
296            .collect())
297    }
298
299    /// Store a failed pre-sign send attempt tombstone.
300    pub async fn add_failed_send_attempt(
301        &self,
302        record: &FailedSendAttemptRecord,
303    ) -> Result<(), Error> {
304        self.put_record(record).await
305    }
306
307    /// List failed pre-sign send attempts for a quote id.
308    pub async fn get_failed_send_attempts_by_quote_id(
309        &self,
310        quote_id: &str,
311    ) -> Result<Vec<FailedSendAttemptRecord>, Error> {
312        let all = self.list_records::<FailedSendAttemptRecord>().await?;
313        Ok(all
314            .into_iter()
315            .filter(|record| record.quote_id == quote_id)
316            .collect())
317    }
318
319    // ── Send Batch storage ───────────────────────────────────────────
320
321    /// Store a new send batch
322    pub async fn store_send_batch(&self, batch: &SendBatchRecord) -> Result<(), Error> {
323        self.put_record(batch).await
324    }
325
326    /// Get a send batch by ID
327    pub async fn get_send_batch(&self, batch_id: &Uuid) -> Result<Option<SendBatchRecord>, Error> {
328        self.get_record::<SendBatchRecord>(&batch_id.to_string())
329            .await
330    }
331
332    /// Update a send batch's state
333    pub async fn update_send_batch(
334        &self,
335        batch_id: &Uuid,
336        new_state: &SendBatchState,
337    ) -> Result<(), Error> {
338        let key = batch_id.to_string();
339        if self.get_send_batch(batch_id).await?.is_none() {
340            return Err(Error::SendBatchNotFound(*batch_id));
341        }
342
343        self.update_record_state::<SendBatchRecord, SendBatchState>(&key, new_state)
344            .await
345    }
346
347    /// Delete a send batch
348    pub async fn delete_send_batch(&self, batch_id: &Uuid) -> Result<(), Error> {
349        self.delete_record::<SendBatchRecord>(&batch_id.to_string())
350            .await
351    }
352
353    /// Get all send batches
354    pub async fn get_all_send_batches(&self) -> Result<Vec<SendBatchRecord>, Error> {
355        self.list_records::<SendBatchRecord>().await
356    }
357
358    // ── Finalized Intent storage (tombstones) ────────────────────────
359
360    /// Look up a finalized intent tombstone by intent ID.
361    pub async fn get_finalized_intent(
362        &self,
363        intent_id: &Uuid,
364    ) -> Result<Option<FinalizedSendIntentRecord>, Error> {
365        self.get_record::<FinalizedSendIntentRecord>(&intent_id.to_string())
366            .await
367    }
368
369    /// Get all finalized send intent tombstones.
370    pub async fn get_all_finalized_send_intents(
371        &self,
372    ) -> Result<Vec<FinalizedSendIntentRecord>, Error> {
373        self.list_records::<FinalizedSendIntentRecord>().await
374    }
375
376    /// Look up a quote ID by the transaction output assigned to a send intent.
377    pub async fn get_quote_id_by_send_outpoint(
378        &self,
379        outpoint: &str,
380    ) -> Result<Option<String>, Error> {
381        let quote_id_bytes = self
382            .kv_store
383            .kv_read(
384                BDK_NAMESPACE,
385                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
386                &outpoint_to_key(outpoint),
387            )
388            .await
389            .map_err(Error::from)?;
390
391        match quote_id_bytes {
392            Some(quote_id_bytes) => String::from_utf8(quote_id_bytes)
393                .map(Some)
394                .map_err(|e| Error::Wallet(format!("Invalid quote-id index entry: {}", e))),
395            None => Ok(None),
396        }
397    }
398
399    /// Populate the send output quote index for records written by older versions.
400    pub(crate) async fn ensure_send_outpoint_quote_id_index(&self) -> Result<(), Error> {
401        if self
402            .kv_store
403            .kv_read(
404                BDK_NAMESPACE,
405                STORAGE_MIGRATION_NAMESPACE,
406                SEND_OUTPOINT_QUOTE_ID_BACKFILL_KEY,
407            )
408            .await
409            .map_err(Error::from)?
410            .is_some()
411        {
412            return Ok(());
413        }
414
415        let (send_intents, finalized_send_intents) = tokio::try_join!(
416            self.get_all_send_intents(),
417            self.get_all_finalized_send_intents(),
418        )?;
419        let mut tx = self
420            .kv_store
421            .begin_transaction()
422            .await
423            .map_err(Error::from)?;
424
425        for intent in send_intents {
426            if let SendIntentState::AwaitingConfirmation { outpoint, .. } = intent.state {
427                tx.kv_write(
428                    BDK_NAMESPACE,
429                    SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
430                    &outpoint_to_key(&outpoint),
431                    intent.quote_id.as_bytes(),
432                )
433                .await
434                .map_err(Error::from)?;
435            }
436        }
437        for intent in finalized_send_intents {
438            tx.kv_write(
439                BDK_NAMESPACE,
440                SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
441                &outpoint_to_key(&intent.outpoint),
442                intent.quote_id.as_bytes(),
443            )
444            .await
445            .map_err(Error::from)?;
446        }
447        tx.kv_write(
448            BDK_NAMESPACE,
449            STORAGE_MIGRATION_NAMESPACE,
450            SEND_OUTPOINT_QUOTE_ID_BACKFILL_KEY,
451            b"complete",
452        )
453        .await
454        .map_err(Error::from)?;
455        tx.commit().await.map_err(Error::from)
456    }
457
458    /// Look up a finalized intent tombstone by quote ID.
459    pub async fn get_finalized_intent_by_quote_id(
460        &self,
461        quote_id: &str,
462    ) -> Result<Option<FinalizedSendIntentRecord>, Error> {
463        let Some(intent_id_bytes) = self
464            .kv_store
465            .kv_read(
466                BDK_NAMESPACE,
467                FINALIZED_SEND_INTENT_QUOTE_ID_NAMESPACE,
468                quote_id,
469            )
470            .await
471            .map_err(Error::from)?
472        else {
473            return Ok(None);
474        };
475
476        let intent_id_str = std::str::from_utf8(&intent_id_bytes)
477            .map_err(|e| Error::Wallet(format!("Invalid intent-id index entry: {}", e)))?;
478        let intent_id = Uuid::from_str(intent_id_str)
479            .map_err(|e| Error::Wallet(format!("Invalid indexed intent id: {}", e)))?;
480
481        self.get_record::<FinalizedSendIntentRecord>(&intent_id.to_string())
482            .await
483    }
484
485    /// Look up a send intent by quote ID.
486    ///
487    /// Scans all active intents and returns the first match.
488    pub async fn get_send_intent_by_quote_id(
489        &self,
490        quote_id: &str,
491    ) -> Result<Option<SendIntentRecord>, Error> {
492        let Some(intent_id_bytes) = self
493            .kv_store
494            .kv_read(BDK_NAMESPACE, SEND_INTENT_QUOTE_ID_NAMESPACE, quote_id)
495            .await
496            .map_err(Error::from)?
497        else {
498            return Ok(None);
499        };
500
501        let intent_id = std::str::from_utf8(&intent_id_bytes)
502            .map_err(|e| Error::Wallet(format!("Invalid quote-id index entry: {}", e)))?;
503        let intent_id = Uuid::from_str(intent_id)
504            .map_err(|e| Error::Wallet(format!("Invalid indexed intent id: {}", e)))?;
505
506        self.get_send_intent(&intent_id).await
507    }
508
509    /// Atomically finalize an active send intent and create a tombstone.
510    pub async fn finalize_send_intent(
511        &self,
512        intent_id: &Uuid,
513        record: &FinalizedSendIntentRecord,
514    ) -> Result<(), Error> {
515        let Some(intent) = self.get_send_intent(intent_id).await? else {
516            return Err(Error::SendIntentNotFound(*intent_id));
517        };
518
519        let serialized = serde_json::to_vec(record)?;
520        let mut tx = self
521            .kv_store
522            .begin_transaction()
523            .await
524            .map_err(Error::from)?;
525        tx.kv_write(
526            BDK_NAMESPACE,
527            FINALIZED_INTENT_NAMESPACE,
528            &record.intent_id.to_string(),
529            &serialized,
530        )
531        .await
532        .map_err(Error::from)?;
533        tx.kv_write(
534            BDK_NAMESPACE,
535            FINALIZED_SEND_INTENT_QUOTE_ID_NAMESPACE,
536            &intent.quote_id,
537            record.intent_id.to_string().as_bytes(),
538        )
539        .await
540        .map_err(Error::from)?;
541        tx.kv_write(
542            BDK_NAMESPACE,
543            SEND_OUTPOINT_QUOTE_ID_NAMESPACE,
544            &outpoint_to_key(&record.outpoint),
545            record.quote_id.as_bytes(),
546        )
547        .await
548        .map_err(Error::from)?;
549        tx.kv_remove(BDK_NAMESPACE, SEND_INTENT_NAMESPACE, &intent_id.to_string())
550            .await
551            .map_err(Error::from)?;
552        tx.kv_remove(
553            BDK_NAMESPACE,
554            SEND_INTENT_QUOTE_ID_NAMESPACE,
555            &intent.quote_id,
556        )
557        .await
558        .map_err(Error::from)?;
559        tx.commit().await.map_err(Error::from)?;
560        Ok(())
561    }
562}
cdk_bdk/storage/send.rs

cdk_bdk/storage/
send.rs
