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

cdk_bdk/storage/
send.rs
