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                }),
164            onchain: settings
165                .onchain
166                .map(|o| cdk_common::payment::OnchainSettings {
167                    confirmations: o.confirmations,
168                    min_receive_amount_sat: o.min_receive_amount_sat,
169                    min_send_amount_sat: o.min_send_amount_sat,
170                }),
171            custom: settings.custom,
172        })
173    }
174
175    /// Create a new invoice
176    async fn create_incoming_payment_request(
177        &self,
178        options: CdkIncomingPaymentOptions,
179    ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
180        let mut inner = self.inner.clone();
181
182        let proto_options = match options {
183            CdkIncomingPaymentOptions::Custom(opts) => IncomingPaymentOptions {
184                options: Some(super::incoming_payment_options::Options::Custom(
185                    super::CustomIncomingPaymentOptions {
186                        description: opts.description,
187                        amount: opts.amount.map(Into::into),
188                        unix_expiry: opts.unix_expiry,
189                        extra_json: opts.extra_json,
190                        quote_id: opts.quote_id.to_string(),
191                        pubkey: opts.pubkey.map(|p| p.to_hex()),
192                    },
193                )),
194            },
195            CdkIncomingPaymentOptions::Bolt11(opts) => IncomingPaymentOptions {
196                options: Some(super::incoming_payment_options::Options::Bolt11(
197                    super::Bolt11IncomingPaymentOptions {
198                        description: opts.description,
199                        amount: Some(opts.amount.into()),
200                        unix_expiry: opts.unix_expiry,
201                    },
202                )),
203            },
204            CdkIncomingPaymentOptions::Bolt12(opts) => IncomingPaymentOptions {
205                options: Some(super::incoming_payment_options::Options::Bolt12(
206                    super::Bolt12IncomingPaymentOptions {
207                        description: opts.description,
208                        amount: opts.amount.map(Into::into),
209                        unix_expiry: opts.unix_expiry,
210                    },
211                )),
212            },
213            CdkIncomingPaymentOptions::Onchain(opts) => IncomingPaymentOptions {
214                options: Some(super::incoming_payment_options::Options::Onchain(
215                    super::OnchainIncomingPaymentOptions {
216                        quote_id: opts.quote_id.to_string(),
217                    },
218                )),
219            },
220        };
221
222        let response = inner
223            .create_payment(Request::new(CreatePaymentRequest {
224                options: Some(proto_options),
225            }))
226            .await
227            .map_err(|err| {
228                tracing::error!("Could not create payment request: {}", err);
229                cdk_common::payment::Error::Custom(err.to_string())
230            })?;
231
232        let response = response.into_inner();
233
234        Ok(response.try_into().map_err(|_| {
235            cdk_common::payment::Error::Anyhow(anyhow!("Could not create create payment response"))
236        })?)
237    }
238
239    async fn get_payment_quote(
240        &self,
241        unit: &cdk_common::CurrencyUnit,
242        options: cdk_common::payment::OutgoingPaymentOptions,
243    ) -> Result<CdkPaymentQuoteResponse, Self::Err> {
244        let mut inner = self.inner.clone();
245
246        let request_type = match &options {
247            cdk_common::payment::OutgoingPaymentOptions::Custom(_) => {
248                OutgoingPaymentRequestType::Custom
249            }
250            cdk_common::payment::OutgoingPaymentOptions::Bolt11(_) => {
251                OutgoingPaymentRequestType::Bolt11Invoice
252            }
253            cdk_common::payment::OutgoingPaymentOptions::Bolt12(_) => {
254                OutgoingPaymentRequestType::Bolt12Offer
255            }
256            cdk_common::payment::OutgoingPaymentOptions::Onchain(_) => {
257                OutgoingPaymentRequestType::Onchain
258            }
259        };
260
261        let proto_request = match &options {
262            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.request.to_string(),
263            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => opts.bolt11.to_string(),
264            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => opts.offer.to_string(),
265            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => opts.address.clone(),
266        };
267
268        let proto_options = match &options {
269            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.melt_options,
270            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => opts.melt_options,
271            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => opts.melt_options,
272            cdk_common::payment::OutgoingPaymentOptions::Onchain(_) => None,
273        };
274
275        let onchain_options = match &options {
276            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => {
277                Some(super::OnchainOutgoingPaymentOptions {
278                    address: opts.address.clone(),
279                    amount: Some(opts.amount.clone().into()),
280                    max_fee_amount: opts.max_fee_amount.clone().into_proto(),
281                    quote_id: opts.quote_id.to_string(),
282                    fee_index: opts.fee_index,
283                    metadata: opts.metadata.clone(),
284                })
285            }
286            _ => None,
287        };
288
289        let extra_json = match &options {
290            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.extra_json.clone(),
291            _ => None,
292        };
293
294        let amount = match &options {
295            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => {
296                opts.amount.clone().into_proto()
297            }
298            _ => None,
299        };
300
301        let quote_id = match &options {
302            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => opts.quote_id.to_string(),
303            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => opts.quote_id.to_string(),
304            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => opts.quote_id.to_string(),
305            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => opts.quote_id.to_string(),
306        };
307
308        let response = inner
309            .get_payment_quote(Request::new(PaymentQuoteRequest {
310                request: proto_request,
311                unit: unit.to_string(),
312                options: proto_options.map(Into::into),
313                request_type: request_type.into(),
314                extra_json,
315                quote_id,
316                onchain_options,
317                amount,
318            }))
319            .await
320            .map_err(|err| {
321                tracing::error!("Could not get payment quote: {}", err);
322                cdk_common::payment::Error::Custom(err.to_string())
323            })?;
324
325        let response = response.into_inner();
326
327        Ok(response.try_into().map_err(|_| {
328            cdk_common::payment::Error::Custom(
329                "Failed to convert payment quote response".to_string(),
330            )
331        })?)
332    }
333
334    async fn make_payment(
335        &self,
336        unit: &cdk_common::CurrencyUnit,
337        options: cdk_common::payment::OutgoingPaymentOptions,
338    ) -> Result<CdkMakePaymentResponse, Self::Err> {
339        let mut inner = self.inner.clone();
340        let payment_options = match options {
341            cdk_common::payment::OutgoingPaymentOptions::Custom(opts) => {
342                super::OutgoingPaymentVariant {
343                    options: Some(super::outgoing_payment_variant::Options::Custom(
344                        super::CustomOutgoingPaymentOptions {
345                            offer: opts.request.to_string(),
346                            amount: opts.amount.map(Into::into),
347                            max_fee_amount: opts.max_fee_amount.into_proto(),
348                            timeout_secs: opts.timeout_secs,
349                            melt_options: opts.melt_options.map(Into::into),
350                            extra_json: opts.extra_json.clone(),
351                            quote_id: opts.quote_id.to_string(),
352                        },
353                    )),
354                }
355            }
356            cdk_common::payment::OutgoingPaymentOptions::Bolt11(opts) => {
357                super::OutgoingPaymentVariant {
358                    options: Some(super::outgoing_payment_variant::Options::Bolt11(
359                        super::Bolt11OutgoingPaymentOptions {
360                            bolt11: opts.bolt11.to_string(),
361                            max_fee_amount: opts.max_fee_amount.into_proto(),
362                            timeout_secs: opts.timeout_secs,
363                            melt_options: opts.melt_options.map(Into::into),
364                            quote_id: opts.quote_id.to_string(),
365                        },
366                    )),
367                }
368            }
369            cdk_common::payment::OutgoingPaymentOptions::Bolt12(opts) => {
370                super::OutgoingPaymentVariant {
371                    options: Some(super::outgoing_payment_variant::Options::Bolt12(
372                        super::Bolt12OutgoingPaymentOptions {
373                            offer: opts.offer.to_string(),
374                            max_fee_amount: opts.max_fee_amount.into_proto(),
375                            timeout_secs: opts.timeout_secs,
376                            melt_options: opts.melt_options.map(Into::into),
377                            quote_id: opts.quote_id.to_string(),
378                        },
379                    )),
380                }
381            }
382            cdk_common::payment::OutgoingPaymentOptions::Onchain(opts) => {
383                super::OutgoingPaymentVariant {
384                    options: Some(super::outgoing_payment_variant::Options::Onchain(
385                        super::OnchainOutgoingPaymentOptions {
386                            address: opts.address.clone(),
387                            amount: Some(opts.amount.into()),
388                            max_fee_amount: opts.max_fee_amount.into_proto(),
389                            quote_id: opts.quote_id.to_string(),
390                            fee_index: opts.fee_index,
391                            metadata: opts.metadata.clone(),
392                        },
393                    )),
394                }
395            }
396        };
397
398        let response = inner
399            .make_payment(Request::new(MakePaymentRequest {
400                payment_options: Some(payment_options),
401                partial_amount: None,
402                max_fee_amount: None,
403                unit: unit.to_string(),
404            }))
405            .await
406            .map_err(|err| {
407                tracing::error!("Could not pay payment request: {}", err);
408
409                if err.message().contains("already paid") {
410                    cdk_common::payment::Error::InvoiceAlreadyPaid
411                } else if err.message().contains("pending") {
412                    cdk_common::payment::Error::InvoicePaymentPending
413                } else {
414                    cdk_common::payment::Error::Custom(err.to_string())
415                }
416            })?;
417
418        let response = response.into_inner();
419
420        Ok(response.try_into().map_err(|_err| {
421            cdk_common::payment::Error::Anyhow(anyhow!("could not make payment"))
422        })?)
423    }
424
425    #[instrument(skip_all)]
426    async fn wait_payment_event(
427        &self,
428    ) -> Result<Pin<Box<dyn Stream<Item = cdk_common::payment::Event> + Send>>, Self::Err> {
429        tracing::debug!("Client waiting for payment");
430        let mut inner = self.inner.clone();
431        let stream = inner
432            .wait_payment_event(Request::new(EmptyRequest {}))
433            .await
434            .map_err(|err| {
435                self.payment_event_stream_is_active
436                    .store(false, Ordering::SeqCst);
437                tracing::error!("Could not open payment event stream: {}", err);
438                cdk_common::payment::Error::Custom(err.to_string())
439            })?
440            .into_inner();
441
442        self.payment_event_stream_is_active
443            .store(true, Ordering::SeqCst);
444
445        let cancel_token = self.cancel_payment_event_stream.lock().await.clone();
446        let cancel_fut = cancel_token.cancelled_owned();
447        let active_flag = self.payment_event_stream_is_active.clone();
448
449        let transformed_stream = stream.take_until(cancel_fut).filter_map(|item| async {
450            match item {
451                Ok(value) => match value.try_into() {
452                    Ok(payment_event) => Some(payment_event),
453                    Err(e) => {
454                        tracing::error!("Error converting payment event: {}", e);
455                        None
456                    }
457                },
458                Err(e) => {
459                    tracing::error!("Error in payment event stream: {}", e);
460                    None
461                }
462            }
463        });
464
465        Ok(Box::pin(ActivePaymentEventStream::new(
466            Box::pin(transformed_stream),
467            active_flag,
468        )))
469    }
470
471    /// Is payment event stream active
472    fn is_payment_event_stream_active(&self) -> bool {
473        self.payment_event_stream_is_active.load(Ordering::SeqCst)
474    }
475
476    /// Cancel payment event stream
477    fn cancel_payment_event_stream(&self) {
478        let cancel_payment_event_stream = Arc::clone(&self.cancel_payment_event_stream);
479
480        tokio::spawn(async move {
481            let mut cancel_token = cancel_payment_event_stream.lock().await;
482            cancel_token.cancel();
483            *cancel_token = CancellationToken::new();
484        });
485    }
486
487    async fn check_incoming_payment_status(
488        &self,
489        payment_identifier: &cdk_common::payment::PaymentIdentifier,
490    ) -> Result<Vec<WaitPaymentResponse>, Self::Err> {
491        let mut inner = self.inner.clone();
492        let response = inner
493            .check_incoming_payment(Request::new(CheckIncomingPaymentRequest {
494                request_identifier: Some(payment_identifier.clone().into()),
495            }))
496            .await
497            .map_err(|err| {
498                tracing::error!("Could not check incoming payment: {}", err);
499                cdk_common::payment::Error::Custom(err.to_string())
500            })?;
501
502        let check_incoming = response.into_inner();
503        check_incoming
504            .payments
505            .into_iter()
506            .map(|resp| resp.try_into().map_err(Self::Err::from))
507            .collect()
508    }
509
510    async fn check_outgoing_payment(
511        &self,
512        payment_identifier: &cdk_common::payment::PaymentIdentifier,
513    ) -> Result<CdkMakePaymentResponse, Self::Err> {
514        let mut inner = self.inner.clone();
515        let response = inner
516            .check_outgoing_payment(Request::new(CheckOutgoingPaymentRequest {
517                request_identifier: Some(payment_identifier.clone().into()),
518            }))
519            .await
520            .map_err(|err| {
521                tracing::error!("Could not check outgoing payment: {}", err);
522                cdk_common::payment::Error::Custom(err.to_string())
523            })?;
524
525        let check_outgoing = response.into_inner();
526
527        Ok(check_outgoing
528            .try_into()
529            .map_err(|_| cdk_common::payment::Error::UnknownPaymentState)?)
530    }
531}
532
533#[cfg(test)]
534mod tests {
535    use std::fs;
536    use std::path::{Path, PathBuf};
537    use std::sync::atomic::{AtomicU64, Ordering};
538
539    use super::load_mtls_config;
540
541    static NEXT_TEST_DIRECTORY: AtomicU64 = AtomicU64::new(0);
542
543    struct TestDirectory(PathBuf);
544
545    impl TestDirectory {
546        fn new() -> Self {
547            let sequence = NEXT_TEST_DIRECTORY.fetch_add(1, Ordering::Relaxed);
548            let path = std::env::temp_dir().join(format!(
549                "cdk-payment-processor-mtls-{}-{sequence}",
550                std::process::id()
551            ));
552            fs::create_dir(&path).expect("create mTLS test directory");
553            Self(path)
554        }
555
556        fn path(&self) -> &Path {
557            &self.0
558        }
559    }
560
561    impl Drop for TestDirectory {
562        fn drop(&mut self) {
563            let _ = fs::remove_dir_all(&self.0);
564        }
565    }
566
567    #[test]
568    fn configured_tls_requires_ca_and_client_identity() {
569        let tls_dir = TestDirectory::new();
570
571        let error = load_mtls_config(tls_dir.path()).expect_err("missing CA should fail");
572        assert!(error.to_string().contains("failed to read CA certificate"));
573
574        fs::write(tls_dir.path().join("ca.pem"), "test CA").expect("write test CA");
575        let error =
576            load_mtls_config(tls_dir.path()).expect_err("missing client certificate should fail");
577        assert!(error
578            .to_string()
579            .contains("failed to read client certificate"));
580
581        fs::write(tls_dir.path().join("client.pem"), "test client certificate")
582            .expect("write test client certificate");
583        let error =
584            load_mtls_config(tls_dir.path()).expect_err("missing client private key should fail");
585        assert!(error
586            .to_string()
587            .contains("failed to read client private key"));
588
589        fs::write(tls_dir.path().join("client.key"), "test client key")
590            .expect("write test client key");
591        load_mtls_config(tls_dir.path()).expect("complete mTLS configuration should load");
592    }
593}
cdk_payment_processor/proto/client.rs

cdk_payment_processor/proto/
client.rs
