cdk_payment_processor/proto/
client.rs

1use std::path::PathBuf;
2use std::pin::Pin;
3use std::str::FromStr;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::Arc;
6
7use anyhow::anyhow;
8use cdk_common::payment::{
9    CreateIncomingPaymentResponse, MakePaymentResponse as CdkMakePaymentResponse, MintPayment,
10    PaymentQuoteResponse,
11};
12use cdk_common::{mint, Amount, CurrencyUnit, MeltOptions, MintQuoteState};
13use futures::{Stream, StreamExt};
14use serde_json::Value;
15use tokio_util::sync::CancellationToken;
16use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
17use tonic::{async_trait, Request};
18use tracing::instrument;
19
20use super::cdk_payment_processor_client::CdkPaymentProcessorClient;
21use super::{
22    CheckIncomingPaymentRequest, CheckOutgoingPaymentRequest, CreatePaymentRequest,
23    MakePaymentRequest, SettingsRequest, WaitIncomingPaymentRequest,
24};
25
26/// Payment Processor
27#[derive(Clone)]
28pub struct PaymentProcessorClient {
29    inner: CdkPaymentProcessorClient<Channel>,
30    wait_incoming_payment_stream_is_active: Arc<AtomicBool>,
31    cancel_incoming_payment_listener: CancellationToken,
32}
33
34impl PaymentProcessorClient {
35    /// Payment Processor
36    pub async fn new(addr: &str, port: u16, tls_dir: Option<PathBuf>) -> anyhow::Result<Self> {
37        let addr = format!("{addr}:{port}");
38        let channel = if let Some(tls_dir) = tls_dir {
39            // TLS directory exists, configure TLS
40
41            // Check for ca.pem
42            let ca_pem_path = tls_dir.join("ca.pem");
43            if !ca_pem_path.exists() {
44                let err_msg = format!("CA certificate file not found: {}", ca_pem_path.display());
45                tracing::error!("{}", err_msg);
46                return Err(anyhow!(err_msg));
47            }
48
49            // Check for client.pem
50            let client_pem_path = tls_dir.join("client.pem");
51            if !client_pem_path.exists() {
52                let err_msg = format!(
53                    "Client certificate file not found: {}",
54                    client_pem_path.display()
55                );
56                tracing::error!("{}", err_msg);
57                return Err(anyhow!(err_msg));
58            }
59
60            // Check for client.key
61            let client_key_path = tls_dir.join("client.key");
62            if !client_key_path.exists() {
63                let err_msg = format!("Client key file not found: {}", client_key_path.display());
64                tracing::error!("{}", err_msg);
65                return Err(anyhow!(err_msg));
66            }
67
68            let server_root_ca_cert = std::fs::read_to_string(&ca_pem_path)?;
69            let server_root_ca_cert = Certificate::from_pem(server_root_ca_cert);
70            let client_cert = std::fs::read_to_string(&client_pem_path)?;
71            let client_key = std::fs::read_to_string(&client_key_path)?;
72            let client_identity = Identity::from_pem(client_cert, client_key);
73            let tls = ClientTlsConfig::new()
74                .ca_certificate(server_root_ca_cert)
75                .identity(client_identity);
76
77            Channel::from_shared(addr)?
78                .tls_config(tls)?
79                .connect()
80                .await?
81        } else {
82            // No TLS directory, skip TLS configuration
83            Channel::from_shared(addr)?.connect().await?
84        };
85
86        let client = CdkPaymentProcessorClient::new(channel);
87
88        Ok(Self {
89            inner: client,
90            wait_incoming_payment_stream_is_active: Arc::new(AtomicBool::new(false)),
91            cancel_incoming_payment_listener: CancellationToken::new(),
92        })
93    }
94}
95
96#[async_trait]
97impl MintPayment for PaymentProcessorClient {
98    type Err = cdk_common::payment::Error;
99
100    async fn get_settings(&self) -> Result<Value, Self::Err> {
101        let mut inner = self.inner.clone();
102        let response = inner
103            .get_settings(Request::new(SettingsRequest {}))
104            .await
105            .map_err(|err| {
106                tracing::error!("Could not get settings: {}", err);
107                cdk_common::payment::Error::Custom(err.to_string())
108            })?;
109
110        let settings = response.into_inner();
111
112        Ok(serde_json::from_str(&settings.inner)?)
113    }
114
115    /// Create a new invoice
116    async fn create_incoming_payment_request(
117        &self,
118        amount: Amount,
119        unit: &CurrencyUnit,
120        description: String,
121        unix_expiry: Option<u64>,
122    ) -> Result<CreateIncomingPaymentResponse, Self::Err> {
123        let mut inner = self.inner.clone();
124        let response = inner
125            .create_payment(Request::new(CreatePaymentRequest {
126                amount: amount.into(),
127                unit: unit.to_string(),
128                description,
129                unix_expiry,
130            }))
131            .await
132            .map_err(|err| {
133                tracing::error!("Could not create payment request: {}", err);
134                cdk_common::payment::Error::Custom(err.to_string())
135            })?;
136
137        let response = response.into_inner();
138
139        Ok(response.try_into().map_err(|_| {
140            cdk_common::payment::Error::Anyhow(anyhow!("Could not create create payment response"))
141        })?)
142    }
143
144    async fn get_payment_quote(
145        &self,
146        request: &str,
147        unit: &CurrencyUnit,
148        options: Option<MeltOptions>,
149    ) -> Result<PaymentQuoteResponse, Self::Err> {
150        let mut inner = self.inner.clone();
151        let response = inner
152            .get_payment_quote(Request::new(super::PaymentQuoteRequest {
153                request: request.to_string(),
154                unit: unit.to_string(),
155                options: options.map(|o| o.into()),
156            }))
157            .await
158            .map_err(|err| {
159                tracing::error!("Could not get payment quote: {}", err);
160                cdk_common::payment::Error::Custom(err.to_string())
161            })?;
162
163        let response = response.into_inner();
164
165        Ok(response.into())
166    }
167
168    async fn make_payment(
169        &self,
170        melt_quote: mint::MeltQuote,
171        partial_amount: Option<Amount>,
172        max_fee_amount: Option<Amount>,
173    ) -> Result<CdkMakePaymentResponse, Self::Err> {
174        let mut inner = self.inner.clone();
175        let response = inner
176            .make_payment(Request::new(MakePaymentRequest {
177                melt_quote: Some(melt_quote.into()),
178                partial_amount: partial_amount.map(|a| a.into()),
179                max_fee_amount: max_fee_amount.map(|a| a.into()),
180            }))
181            .await
182            .map_err(|err| {
183                tracing::error!("Could not pay payment request: {}", err);
184
185                if err.message().contains("already paid") {
186                    cdk_common::payment::Error::InvoiceAlreadyPaid
187                } else if err.message().contains("pending") {
188                    cdk_common::payment::Error::InvoicePaymentPending
189                } else {
190                    cdk_common::payment::Error::Custom(err.to_string())
191                }
192            })?;
193
194        let response = response.into_inner();
195
196        Ok(response.try_into().map_err(|_err| {
197            cdk_common::payment::Error::Anyhow(anyhow!("could not make payment"))
198        })?)
199    }
200
201    /// Listen for invoices to be paid to the mint
202    #[instrument(skip_all)]
203    async fn wait_any_incoming_payment(
204        &self,
205    ) -> Result<Pin<Box<dyn Stream<Item = String> + Send>>, Self::Err> {
206        self.wait_incoming_payment_stream_is_active
207            .store(true, Ordering::SeqCst);
208        tracing::debug!("Client waiting for payment");
209        let mut inner = self.inner.clone();
210        let stream = inner
211            .wait_incoming_payment(WaitIncomingPaymentRequest {})
212            .await
213            .map_err(|err| {
214                tracing::error!("Could not check incoming payment stream: {}", err);
215                cdk_common::payment::Error::Custom(err.to_string())
216            })?
217            .into_inner();
218
219        let cancel_token = self.cancel_incoming_payment_listener.clone();
220        let cancel_fut = cancel_token.cancelled_owned();
221        let active_flag = self.wait_incoming_payment_stream_is_active.clone();
222
223        let transformed_stream = stream
224            .take_until(cancel_fut)
225            .filter_map(|item| async move {
226                match item {
227                    Ok(value) => {
228                        tracing::warn!("{}", value.lookup_id);
229                        Some(value.lookup_id)
230                    }
231                    Err(e) => {
232                        tracing::error!("Error in payment stream: {}", e);
233                        None // Skip this item and continue with the stream
234                    }
235                }
236            })
237            .inspect(move |_| {
238                active_flag.store(false, Ordering::SeqCst);
239                tracing::info!("Payment stream inactive");
240            });
241
242        Ok(Box::pin(transformed_stream))
243    }
244
245    /// Is wait invoice active
246    fn is_wait_invoice_active(&self) -> bool {
247        self.wait_incoming_payment_stream_is_active
248            .load(Ordering::SeqCst)
249    }
250
251    /// Cancel wait invoice
252    fn cancel_wait_invoice(&self) {
253        self.cancel_incoming_payment_listener.cancel();
254    }
255
256    async fn check_incoming_payment_status(
257        &self,
258        request_lookup_id: &str,
259    ) -> Result<MintQuoteState, Self::Err> {
260        let mut inner = self.inner.clone();
261        let response = inner
262            .check_incoming_payment(Request::new(CheckIncomingPaymentRequest {
263                request_lookup_id: request_lookup_id.to_string(),
264            }))
265            .await
266            .map_err(|err| {
267                tracing::error!("Could not check incoming payment: {}", err);
268                cdk_common::payment::Error::Custom(err.to_string())
269            })?;
270
271        let check_incoming = response.into_inner();
272
273        let status = check_incoming.status().as_str_name();
274
275        Ok(MintQuoteState::from_str(status)?)
276    }
277
278    async fn check_outgoing_payment(
279        &self,
280        request_lookup_id: &str,
281    ) -> Result<CdkMakePaymentResponse, Self::Err> {
282        let mut inner = self.inner.clone();
283        let response = inner
284            .check_outgoing_payment(Request::new(CheckOutgoingPaymentRequest {
285                request_lookup_id: request_lookup_id.to_string(),
286            }))
287            .await
288            .map_err(|err| {
289                tracing::error!("Could not check outgoing payment: {}", err);
290                cdk_common::payment::Error::Custom(err.to_string())
291            })?;
292
293        let check_outgoing = response.into_inner();
294
295        Ok(check_outgoing
296            .try_into()
297            .map_err(|_| cdk_common::payment::Error::UnknownPaymentState)?)
298    }
299}
cdk_payment_processor/proto/client.rs

cdk_payment_processor/proto/
client.rs
