cdk_payment_processor/proto/
client.rs

1use std::path::{Path, PathBuf};
2use std::pin::Pin;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::Arc;
5use std::task::{Context, Poll};
6
7use anyhow::{anyhow, Context as AnyhowContext};
8use cdk_common::grpc::{VersionInterceptor, VERSION_HEADER};
9use cdk_common::payment::{
10    CreateIncomingPaymentResponse, IncomingPaymentOptions as CdkIncomingPaymentOptions,
11    MakePaymentResponse as CdkMakePaymentResponse, MintPayment,
12    PaymentQuoteResponse as CdkPaymentQuoteResponse, WaitPaymentResponse,
13};
14use futures::{Stream, StreamExt};
15use tokio::sync::Mutex;
16use tokio_util::sync::CancellationToken;
17use tonic::codegen::InterceptedService;
18use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
19use tonic::{async_trait, Request};
20use tracing::instrument;
21
22use crate::proto::cdk_payment_processor_client::CdkPaymentProcessorClient;
23use crate::proto::{
24    CheckIncomingPaymentRequest, CheckOutgoingPaymentRequest, CreatePaymentRequest, EmptyRequest,
25    IncomingPaymentOptions, IntoProtoAmount, MakePaymentRequest, OutgoingPaymentRequestType,
26    PaymentQuoteRequest,
27};
28
29/// Payment Processor
30#[derive(Clone)]
31pub struct PaymentProcessorClient {
32    inner: CdkPaymentProcessorClient<InterceptedService<Channel, VersionInterceptor>>,
33    payment_event_stream_is_active: Arc<AtomicBool>,
34    cancel_payment_event_stream: Arc<Mutex<CancellationToken>>,
35}
36
37struct ActivePaymentEventStream {
38    inner: Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>,
39    active_flag: Arc<AtomicBool>,
40}
41
42impl ActivePaymentEventStream {
43    fn new(
44        inner: Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>,
45        active_flag: Arc<AtomicBool>,
46    ) -> Self {
47        Self { inner, active_flag }
48    }
49}
50
51impl Drop for ActivePaymentEventStream {
52    fn drop(&mut self) {
53        self.active_flag.store(false, Ordering::SeqCst);
54        tracing::info!("Payment event stream inactive");
55    }
56}
57
58impl Stream for ActivePaymentEventStream {
59    type Item = cdk_common::payment::Event;
60
61    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
62        let this = self.get_mut();
63        this.inner.as_mut().poll_next(cx)
64    }
65}
66
67impl std::fmt::Debug for PaymentProcessorClient {
68    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
69        f.debug_struct("PaymentProcessorClient")
70            .finish_non_exhaustive()
71    }
72}
73
74impl PaymentProcessorClient {
75    /// Connect to a payment processor.
76    ///
77    /// When `tls_dir` is provided, it must contain `ca.pem`, `client.pem`, and
78    /// `client.key`. The CA certificate authenticates the server, while the
79    /// client certificate and key authenticate this client to the server.
80    pub async fn new(addr: &str, port: u16, tls_dir: Option<PathBuf>) -> anyhow::Result<Self> {
81        let scheme = if tls_dir.is_some() { "https" } else { "http" };
82        let endpoint = format!("{scheme}://{addr}:{port}");
83
84        let channel = if let Some(tls_dir) = tls_dir {
85            let tls = load_mtls_config(&tls_dir)?;
86            Channel::from_shared(endpoint)?
87                .tls_config(tls)?
88                .connect()
89                .await?
90        } else {
91            // No TLS directory, skip TLS configuration
92            Channel::from_shared(endpoint)?.connect().await?
93        };
94
95        let interceptor = VersionInterceptor::new(
96            VERSION_HEADER,
97            cdk_common::PAYMENT_PROCESSOR_PROTOCOL_VERSION,
98        );
99        let client = CdkPaymentProcessorClient::with_interceptor(channel, interceptor);
100
101        Ok(Self {
102            inner: client,
103            payment_event_stream_is_active: Arc::new(AtomicBool::new(false)),
104            cancel_payment_event_stream: Arc::new(Mutex::new(CancellationToken::new())),
105        })
106    }
107}
108
109fn load_mtls_config(tls_dir: &Path) -> anyhow::Result<ClientTlsConfig> {
110    let ca_pem_path = tls_dir.join("ca.pem");
111    let client_pem_path = tls_dir.join("client.pem");
112    let client_key_path = tls_dir.join("client.key");
113
114    let server_root_ca_cert = std::fs::read(&ca_pem_path)
115        .with_context(|| format!("failed to read CA certificate `{}`", ca_pem_path.display()))?;
116    let client_cert = std::fs::read(&client_pem_path).with_context(|| {
117        format!(
118            "failed to read client certificate `{}`",
119            client_pem_path.display()
120        )
121    })?;
122    let client_key = std::fs::read(&client_key_path).with_context(|| {
123        format!(
124            "failed to read client private key `{}`",
125            client_key_path.display()
126        )
127    })?;
128
129    Ok(ClientTlsConfig::new()
130        .ca_certificate(Certificate::from_pem(server_root_ca_cert))
131        .identity(Identity::from_pem(client_cert, client_key)))
132}
133
134#[async_trait]
135impl MintPayment for PaymentProcessorClient {
136    type Err = cdk_common::payment::Error;
137
138    async fn get_settings(&self) -> Result<cdk_common::payment::SettingsResponse, Self::Err> {
139        let mut inner = self.inner.clone();
140        let response = inner
141            .get_settings(Request::new(EmptyRequest {}))
142            .await
143            .map_err(|err| {
144                tracing::error!("Could not get settings: {}", err);
145                cdk_common::payment::Error::Custom(err.to_string())
146            })?;
147
148        let settings = response.into_inner();
149
150        Ok(cdk_common::payment::SettingsResponse {
151            unit: settings.unit,
152            bolt11: settings
153                .bolt11
154                .map(|b| cdk_common::payment::Bolt11Settings {
155                    mpp: b.mpp,
156                    amountless: b.amountless,
157                    invoice_description: b.invoice_description,
158                }),
159            bolt12: settings
160                .bolt12
161                .map(|b| cdk_common::payment::Bolt12Settings {
162                    amountless: b.amountless,
163                    invoice_description: b.invoice_description,
164                }),
165            onchain: settings
166                .onchain
167                .map(|o| cdk_common::payment::OnchainSettings {
168                    confirmations: o.confirmations,
169                    min_receive_amount_sat: o.min_receive_amount_sat,
170                    min_send_amount_sat: o.min_send_amount_sat,
171                }),
172            custom: settings.custom,
173        })
174    }
175
176    /// Create a new invoice
177    async fn create_incoming_payment_request(
178        &self,
179        options: CdkIncomingPaymentOptions,
180    ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
181        let mut inner = self.inner.clone();
182
183        let proto_options = match options {
184            CdkIncomingPaymentOptions::Custom(opts) => IncomingPaymentOptions {
185                options: Some(super::incoming_payment_options::Options::Custom(
186                    super::CustomIncomingPaymentOptions {
187                        description: opts.description,
188                        amount: opts.amount.map(Into::into),
189                        unix_expiry: opts.unix_expiry,
190                        extra_json: opts.extra_json,
191                        quote_id: opts.quote_id.to_string(),
192                        pubkey: opts.pubkey.map(|p| p.to_hex()),
193                    },
194                )),
195            },
196            CdkIncomingPaymentOptions::Bolt11(opts) => IncomingPaymentOptions {
197                options: Some(super::incoming_payment_options::Options::Bolt11(
198                    super::Bolt11IncomingPaymentOptions {
199                        description: opts.description,
200                        amount: Some(opts.amount.into()),
201                        unix_expiry: opts.unix_expiry,
202                    },
203                )),
204            },
205            CdkIncomingPaymentOptions::Bolt12(opts) => IncomingPaymentOptions {
206                options: Some(super::incoming_payment_options::Options::Bolt12(
207                    super::Bolt12IncomingPaymentOptions {
208                        description: opts.description,
209                        amount: opts.amount.map(Into::into),
210                        unix_expiry: opts.unix_expiry,
211                    },
212                )),
213            },
214            CdkIncomingPaymentOptions::Onchain(opts) => IncomingPaymentOptions {
215                options: Some(super::incoming_payment_options::Options::Onchain(
216                    super::OnchainIncomingPaymentOptions {
217                        quote_id: opts.quote_id.to_string(),
218                    },
219                )),
220            },
221        };
222
223        let response = inner
224            .create_payment(Request::new(CreatePaymentRequest {
225                options: Some(proto_options),
226            }))
227            .await
228            .map_err(|err| {
229                tracing::error!("Could not create payment request: {}", err);
230                cdk_common::payment::Error::Custom(err.to_string())
231            })?;
232
233        let response = response.into_inner();
234
235        Ok(response.try_into().map_err(|_| {
236            cdk_common::payment::Error::Anyhow(anyhow!("Could not create create payment response"))
237        })?)
238    }
239
240    async fn get_payment_quote(
241        &self,
242        unit: &cdk_common::CurrencyUnit,
243        options: cdk_common::payment::OutgoingPaymentOptions,
244    ) -> Result<CdkPaymentQuoteResponse, Self::Err> {
245        let mut inner = self.inner.clone();
246
247        let request_type = match &options {
248            cdk_common::payment::OutgoingPaymentOptions::Custom(_) => {
249                OutgoingPaymentRequestType::Custom
250            }
251            cdk_common::payment::OutgoingPaymentOptions::Bolt11(_) => {
252                OutgoingPaymentRequestType::Bolt11Invoice
253            }
254            cdk_common::payment::OutgoingPaymentOptions::Bolt12(_) => {
255                OutgoingPaymentRequestType::Bolt12Offer
256            }
257            cdk_common::payment::OutgoingPaymentOptions::Onchain(_) => {
258                OutgoingPaymentRequestType::Onchain
259            }
260        };
261
262        let proto_request = match &options {
263            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.request.to_string(),
264            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => opts.bolt11.to_string(),
265            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => opts.offer.to_string(),
266            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => opts.address.clone(),
267        };
268
269        let proto_options = match &options {
270            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.melt_options,
271            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => opts.melt_options,
272            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => opts.melt_options,
273            cdk_common::payment::OutgoingPaymentOptions::Onchain(_) => None,
274        };
275
276        let onchain_options = match &options {
277            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => {
278                Some(super::OnchainOutgoingPaymentOptions {
279                    address: opts.address.clone(),
280                    amount: Some(opts.amount.clone().into()),
281                    max_fee_amount: opts.max_fee_amount.clone().into_proto(),
282                    quote_id: opts.quote_id.to_string(),
283                    fee_index: opts.fee_index,
284                    metadata: opts.metadata.clone(),
285                })
286            }
287            _ => None,
288        };
289
290        let extra_json = match &options {
291            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.extra_json.clone(),
292            _ => None,
293        };
294
295        let amount = match &options {
296            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => {
297                opts.amount.clone().into_proto()
298            }
299            _ => None,
300        };
301
302        let quote_id = match &options {
303            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.quote_id.to_string(),
304            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => opts.quote_id.to_string(),
305            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => opts.quote_id.to_string(),
306            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => opts.quote_id.to_string(),
307        };
308
309        let response = inner
310            .get_payment_quote(Request::new(PaymentQuoteRequest {
311                request: proto_request,
312                unit: unit.to_string(),
313                options: proto_options.map(Into::into),
314                request_type: request_type.into(),
315                extra_json,
316                quote_id,
317                onchain_options,
318                amount,
319            }))
320            .await
321            .map_err(|err| {
322                tracing::error!("Could not get payment quote: {}", err);
323                cdk_common::payment::Error::Custom(err.to_string())
324            })?;
325
326        let response = response.into_inner();
327
328        Ok(response.try_into().map_err(|_| {
329            cdk_common::payment::Error::Custom(
330                "Failed to convert payment quote response".to_string(),
331            )
332        })?)
333    }
334
335    async fn make_payment(
336        &self,
337        unit: &cdk_common::CurrencyUnit,
338        options: cdk_common::payment::OutgoingPaymentOptions,
339    ) -> Result<CdkMakePaymentResponse, Self::Err> {
340        let mut inner = self.inner.clone();
341        let payment_options = match options {
342            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => {
343                super::OutgoingPaymentVariant {
344                    options: Some(super::outgoing_payment_variant::Options::Custom(
345                        super::CustomOutgoingPaymentOptions {
346                            offer: opts.request.to_string(),
347                            amount: opts.amount.map(Into::into),
348                            max_fee_amount: opts.max_fee_amount.into_proto(),
349                            timeout_secs: opts.timeout_secs,
350                            melt_options: opts.melt_options.map(Into::into),
351                            extra_json: opts.extra_json.clone(),
352                            quote_id: opts.quote_id.to_string(),
353                        },
354                    )),
355                }
356            }
357            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => {
358                super::OutgoingPaymentVariant {
359                    options: Some(super::outgoing_payment_variant::Options::Bolt11(
360                        super::Bolt11OutgoingPaymentOptions {
361                            bolt11: opts.bolt11.to_string(),
362                            max_fee_amount: opts.max_fee_amount.into_proto(),
363                            timeout_secs: opts.timeout_secs,
364                            melt_options: opts.melt_options.map(Into::into),
365                            quote_id: opts.quote_id.to_string(),
366                        },
367                    )),
368                }
369            }
370            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => {
371                super::OutgoingPaymentVariant {
372                    options: Some(super::outgoing_payment_variant::Options::Bolt12(
373                        super::Bolt12OutgoingPaymentOptions {
374                            offer: opts.offer.to_string(),
375                            max_fee_amount: opts.max_fee_amount.into_proto(),
376                            timeout_secs: opts.timeout_secs,
377                            melt_options: opts.melt_options.map(Into::into),
378                            quote_id: opts.quote_id.to_string(),
379                        },
380                    )),
381                }
382            }
383            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => {
384                super::OutgoingPaymentVariant {
385                    options: Some(super::outgoing_payment_variant::Options::Onchain(
386                        super::OnchainOutgoingPaymentOptions {
387                            address: opts.address.clone(),
388                            amount: Some(opts.amount.into()),
389                            max_fee_amount: opts.max_fee_amount.into_proto(),
390                            quote_id: opts.quote_id.to_string(),
391                            fee_index: opts.fee_index,
392                            metadata: opts.metadata.clone(),
393                        },
394                    )),
395                }
396            }
397        };
398
399        let response = inner
400            .make_payment(Request::new(MakePaymentRequest {
401                payment_options: Some(payment_options),
402                partial_amount: None,
403                max_fee_amount: None,
404                unit: unit.to_string(),
405            }))
406            .await
407            .map_err(|err| {
408                tracing::error!("Could not pay payment request: {}", err);
409
410                if err.message().contains("already paid") {
411                    cdk_common::payment::Error::InvoiceAlreadyPaid
412                } else if err.message().contains("pending") {
413                    cdk_common::payment::Error::InvoicePaymentPending
414                } else {
415                    cdk_common::payment::Error::Custom(err.to_string())
416                }
417            })?;
418
419        let response = response.into_inner();
420
421        Ok(response.try_into().map_err(|_err| {
422            cdk_common::payment::Error::Anyhow(anyhow!("could not make payment"))
423        })?)
424    }
425
426    #[instrument(skip_all)]
427    async fn wait_payment_event(
428        &self,
429    ) -> Result<Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>, Self::Err> {
430        tracing::debug!("Client waiting for payment");
431        let mut inner = self.inner.clone();
432        let stream = inner
433            .wait_payment_event(Request::new(EmptyRequest {}))
434            .await
435            .map_err(|err| {
436                self.payment_event_stream_is_active
437                    .store(false, Ordering::SeqCst);
438                tracing::error!("Could not open payment event stream: {}", err);
439                cdk_common::payment::Error::Custom(err.to_string())
440            })?
441            .into_inner();
442
443        self.payment_event_stream_is_active
444            .store(true, Ordering::SeqCst);
445
446        let cancel_token = self.cancel_payment_event_stream.lock().await.clone();
447        let cancel_fut = cancel_token.cancelled_owned();
448        let active_flag = self.payment_event_stream_is_active.clone();
449
450        let transformed_stream = stream.take_until(cancel_fut).filter_map(|item| async {
451            match item {
452                Ok(value) => match value.try_into() {
453                    Ok(payment_event) => Some(payment_event),
454                    Err(e) => {
455                        tracing::error!("Error converting payment event: {}", e);
456                        None
457                    }
458                },
459                Err(e) => {
460                    tracing::error!("Error in payment event stream: {}", e);
461                    None
462                }
463            }
464        });
465
466        Ok(Box::pin(ActivePaymentEventStream::new(
467            Box::pin(transformed_stream),
468            active_flag,
469        )))
470    }
471
472    /// Is payment event stream active
473    fn is_payment_event_stream_active(&self) -> bool {
474        self.payment_event_stream_is_active.load(Ordering::SeqCst)
475    }
476
477    /// Cancel payment event stream
478    fn cancel_payment_event_stream(&self) {
479        let cancel_payment_event_stream = Arc::clone(&self.cancel_payment_event_stream);
480
481        tokio::spawn(async move {
482            let mut cancel_token = cancel_payment_event_stream.lock().await;
483            cancel_token.cancel();
484            *cancel_token = CancellationToken::new();
485        });
486    }
487
488    async fn check_incoming_payment_status(
489        &self,
490        payment_identifier: &cdk_common::payment::PaymentIdentifier,
491    ) -> Result<Vec<WaitPaymentResponse>, Self::Err> {
492        let mut inner = self.inner.clone();
493        let response = inner
494            .check_incoming_payment(Request::new(CheckIncomingPaymentRequest {
495                request_identifier: Some(payment_identifier.clone().into()),
496            }))
497            .await
498            .map_err(|err| {
499                tracing::error!("Could not check incoming payment: {}", err);
500                cdk_common::payment::Error::Custom(err.to_string())
501            })?;
502
503        let check_incoming = response.into_inner();
504        check_incoming
505            .payments
506            .into_iter()
507            .map(|resp| resp.try_into().map_err(Self::Err::from))
508            .collect()
509    }
510
511    async fn check_outgoing_payment(
512        &self,
513        payment_identifier: &cdk_common::payment::PaymentIdentifier,
514    ) -> Result<CdkMakePaymentResponse, Self::Err> {
515        let mut inner = self.inner.clone();
516        let response = inner
517            .check_outgoing_payment(Request::new(CheckOutgoingPaymentRequest {
518                request_identifier: Some(payment_identifier.clone().into()),
519            }))
520            .await
521            .map_err(|err| {
522                tracing::error!("Could not check outgoing payment: {}", err);
523                cdk_common::payment::Error::Custom(err.to_string())
524            })?;
525
526        let check_outgoing = response.into_inner();
527
528        Ok(check_outgoing
529            .try_into()
530            .map_err(|_| cdk_common::payment::Error::UnknownPaymentState)?)
531    }
532}
533
534#[cfg(test)]
535mod tests {
536    use std::fs;
537    use std::path::{Path, PathBuf};
538    use std::sync::atomic::{AtomicU64, Ordering};
539
540    use super::load_mtls_config;
541
542    static NEXT_TEST_DIRECTORY: AtomicU64 = AtomicU64::new(0);
543
544    struct TestDirectory(PathBuf);
545
546    impl TestDirectory {
547        fn new() -> Self {
548            let sequence = NEXT_TEST_DIRECTORY.fetch_add(1, Ordering::Relaxed);
549            let path = std::env::temp_dir().join(format!(
550                "cdk-payment-processor-mtls-{}-{sequence}",
551                std::process::id()
552            ));
553            fs::create_dir(&path).expect("create mTLS test directory");
554            Self(path)
555        }
556
557        fn path(&self) -> &Path {
558            &self.0
559        }
560    }
561
562    impl Drop for TestDirectory {
563        fn drop(&mut self) {
564            let _ = fs::remove_dir_all(&self.0);
565        }
566    }
567
568    #[test]
569    fn configured_tls_requires_ca_and_client_identity() {
570        let tls_dir = TestDirectory::new();
571
572        let error = load_mtls_config(tls_dir.path()).expect_err("missing CA should fail");
573        assert!(error.to_string().contains("failed to read CA certificate"));
574
575        fs::write(tls_dir.path().join("ca.pem"), "test CA").expect("write test CA");
576        let error =
577            load_mtls_config(tls_dir.path()).expect_err("missing client certificate should fail");
578        assert!(error
579            .to_string()
580            .contains("failed to read client certificate"));
581
582        fs::write(tls_dir.path().join("client.pem"), "test client certificate")
583            .expect("write test client certificate");
584        let error =
585            load_mtls_config(tls_dir.path()).expect_err("missing client private key should fail");
586        assert!(error
587            .to_string()
588            .contains("failed to read client private key"));
589
590        fs::write(tls_dir.path().join("client.key"), "test client key")
591            .expect("write test client key");
592        load_mtls_config(tls_dir.path()).expect("complete mTLS configuration should load");
593    }
594}
cdk_payment_processor/proto/client.rs

cdk_payment_processor/proto/
client.rs
