acta_maker_sdk/ws/vault/
client.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use tokio::sync::{broadcast, watch};
5use uuid::Uuid;
6
7use crate::orders::SignerLike;
8use crate::quote::{QuoteBuildError, QuoteBuilder};
9use crate::types::VaultDelegatePermissions;
10use crate::types::ids::OrderId;
11use crate::wire::{WireError, decode_base58_32};
12use crate::ws::managed::{
13    ManagedMessageReceiver, ManagedWsConfigError, ManagedWsError, ManagedWsEvent, ManagedWsHandle,
14    ManagedWsState, ReferenceChangeReceiver, SendAwaitError, SendTicket, WaitUntilReadyError,
15    spawn_managed_ws,
16};
17use crate::ws::types::{
18    AcceptQuoteMessage, BatchQuotesMessage, CancelAllQuotesMessage, CancelQuoteData, CancelRfqData,
19    ClientMessage, Delivery, GetSwapStatusMessage, QuoteMessage, ReplaceQuoteMessage,
20    RfqRequestMessage, ServerError, ServerMessage, SponsoredTxToSignData,
21    SubmitSignedSponsoredTxData, SubscribeData, SwapQuoteData, SwapQuoteResultData,
22    SwapStatusMessage, SwapSubmitData, SwapSubmitResultData, UnsubscribeData, VaultAuthSuccessData,
23};
24
25use solana_sdk::message::AddressLookupTableAccount;
26
27use crate::chain::vault::{
28    ComputeBudgetBounds, VaultOpenIntent, VaultSwapIntent, verify_vault_open, verify_vault_swap,
29};
30
31use super::config::VaultOperatorConfig;
32use super::gate::SendGate;
33use super::projection::VaultProjectionReceiver;
34use super::tx::{VaultTxError, ensure_deadline_ahead, ensure_system_time_ahead};
35
36#[derive(Debug, thiserror::Error)]
37#[non_exhaustive]
38pub enum VaultClientError {
39    #[error("vault_pda is not a base58 32-byte address")]
40    InvalidVaultPda(#[source] WireError),
41    #[error(transparent)]
42    InvalidConfig(#[from] ManagedWsConfigError),
43}
44
45#[derive(Debug, thiserror::Error)]
46#[non_exhaustive]
47pub enum VaultRequestError {
48    #[error(transparent)]
49    Session(#[from] ManagedWsError),
50    #[error(transparent)]
51    Transport(#[from] SendAwaitError),
52    #[error("server rejected request {request_id}")]
53    Rejected {
54        request_id: Uuid,
55        response: Arc<ServerMessage>,
56    },
57    #[error("expected {expected}, received a different correlated response")]
58    UnexpectedResponse {
59        expected: &'static str,
60        response: Arc<ServerMessage>,
61    },
62    #[error("the send gate refused the command")]
63    Gated,
64}
65
66impl VaultRequestError {
67    #[must_use]
68    pub fn server_error(&self) -> Option<&ServerError> {
69        match self {
70            Self::Rejected { response, .. } => match &**response {
71                ServerMessage::RequestError(error) => Some(&error.error),
72                _ => None,
73            },
74            _ => None,
75        }
76    }
77
78    /// Whether the command can have reached the server; `None` is a rejection
79    /// of a command without delivery semantics, which did not execute.
80    #[must_use]
81    pub fn delivery(&self) -> Option<Delivery> {
82        match self {
83            Self::Rejected { response, .. } => match &**response {
84                ServerMessage::RequestError(error) => error.delivery,
85                _ => Some(Delivery::Unknown),
86            },
87            Self::Session(ManagedWsError::NotReady | ManagedWsError::QueueFull)
88            | Self::Gated
89            | Self::Transport(
90                SendAwaitError::SubscriptionPending
91                | SendAwaitError::NotReady
92                | SendAwaitError::NoCorrelationKey
93                | SendAwaitError::DuplicateInFlight
94                | SendAwaitError::TooManyPending { .. }
95                | SendAwaitError::QueueFull
96                | SendAwaitError::InvalidMessage(_),
97            ) => Some(Delivery::NotForwarded),
98            Self::Session(_) | Self::Transport(_) | Self::UnexpectedResponse { .. } => {
99                Some(Delivery::Unknown)
100            }
101        }
102    }
103}
104
105/// Vault operator session on `/vault`; every reconnect re-runs `StartVaultAuth`.
106#[derive(Clone)]
107pub struct VaultOperatorClient {
108    inner: ManagedWsHandle,
109    vault_pda: String,
110    vault_owner: [u8; 32],
111    operator: Arc<dyn SignerLike + Send + Sync>,
112    request_timeout: Duration,
113}
114
115impl VaultOperatorClient {
116    pub fn spawn(config: VaultOperatorConfig) -> Result<Self, VaultClientError> {
117        let vault_owner =
118            decode_base58_32(&config.vault_pda).map_err(VaultClientError::InvalidVaultPda)?;
119        Ok(Self {
120            inner: spawn_managed_ws(config.managed)?,
121            vault_pda: config.vault_pda,
122            vault_owner,
123            operator: config.operator,
124            request_timeout: config.request_timeout,
125        })
126    }
127
128    #[must_use]
129    pub fn vault_pda(&self) -> &str {
130        &self.vault_pda
131    }
132
133    /// Vault address used as `maker` in the order preimage.
134    #[must_use]
135    pub const fn vault_owner(&self) -> [u8; 32] {
136        self.vault_owner
137    }
138
139    #[must_use]
140    pub fn operator_pubkey(&self) -> String {
141        self.operator.pubkey_base58()
142    }
143
144    /// `None` before authentication and after it ends.
145    #[must_use]
146    pub fn auth(&self) -> Option<VaultAuthSuccessData> {
147        self.inner.vault_auth()
148    }
149
150    #[must_use]
151    pub fn permissions(&self) -> Option<VaultDelegatePermissions> {
152        self.auth().map(|auth| auth.permissions)
153    }
154
155    #[must_use]
156    pub fn can_maker(&self) -> bool {
157        self.inner.is_ready()
158            && self.auth().is_some_and(|auth| {
159                matches!(
160                    auth.operator_kind,
161                    crate::types::VaultOperatorKind::Delegate
162                        | crate::types::VaultOperatorKind::Governance
163                ) && auth.permissions.can_maker()
164            })
165    }
166
167    #[must_use]
168    pub fn can_taker(&self) -> bool {
169        self.inner.is_ready()
170            && self
171                .permissions()
172                .is_some_and(VaultDelegatePermissions::can_taker)
173    }
174
175    #[must_use]
176    pub fn subscribe_vault_auth(&self) -> watch::Receiver<Option<VaultAuthSuccessData>> {
177        self.inner.subscribe_vault_auth()
178    }
179
180    pub fn sign_quote(&self, quote: &QuoteBuilder<'_>) -> Result<QuoteMessage, QuoteBuildError> {
181        quote
182            .clone()
183            .maker_owner(self.vault_owner)
184            .sign(self.operator.as_ref())
185    }
186
187    /// Send only in `connection_epoch`, the epoch whose recovery the strategy
188    /// applied; a reconnect in between rejects the quote with `NotReady`.
189    pub async fn quote_in_epoch(
190        &self,
191        payload: QuoteMessage,
192        connection_epoch: u64,
193    ) -> Result<(), ManagedWsError> {
194        self.inner
195            .send_in_epoch(ClientMessage::Quote(payload), connection_epoch)
196            .await
197    }
198
199    pub fn try_quote_in_epoch(
200        &self,
201        payload: QuoteMessage,
202        connection_epoch: u64,
203    ) -> Result<SendTicket, ManagedWsError> {
204        self.inner
205            .try_send_in_epoch(ClientMessage::Quote(payload), connection_epoch)
206    }
207
208    pub async fn replace_in_epoch(
209        &self,
210        payload: ReplaceQuoteMessage,
211        connection_epoch: u64,
212    ) -> Result<(), ManagedWsError> {
213        self.inner
214            .send_in_epoch(ClientMessage::ReplaceQuote(payload), connection_epoch)
215            .await
216    }
217
218    pub async fn batch_quotes_in_epoch(
219        &self,
220        payload: BatchQuotesMessage,
221        connection_epoch: u64,
222    ) -> Result<(), ManagedWsError> {
223        self.inner
224            .send_in_epoch(ClientMessage::BatchQuotes(payload), connection_epoch)
225            .await
226    }
227
228    pub async fn cancel_quote(&self, payload: CancelQuoteData) -> Result<(), ManagedWsError> {
229        self.send(ClientMessage::CancelQuote(payload)).await
230    }
231
232    pub async fn cancel_all_quotes(
233        &self,
234        payload: CancelAllQuotesMessage,
235    ) -> Result<(), ManagedWsError> {
236        self.send(ClientMessage::CancelAllQuotes(payload)).await
237    }
238
239    /// Resolves on the server's receipt; `RfqCreated` still arrives as a push.
240    pub async fn rfq_request(&self, payload: RfqRequestMessage) -> Result<(), VaultRequestError> {
241        self.command(ClientMessage::RfqRequest(payload)).await
242    }
243
244    pub async fn cancel_rfq(&self, payload: CancelRfqData) -> Result<(), ManagedWsError> {
245        self.send(ClientMessage::CancelRfq(payload)).await
246    }
247
248    pub async fn accept_quote(&self, payload: AcceptQuoteMessage) -> Result<(), VaultRequestError> {
249        self.command(ClientMessage::AcceptQuote(payload)).await
250    }
251
252    /// Signs only if the served transaction opens exactly `intent` within `bounds`.
253    pub fn verify_and_sign_sponsored_tx(
254        &self,
255        to_sign: &SponsoredTxToSignData,
256        intent: &VaultOpenIntent<'_>,
257        tables: &[AddressLookupTableAccount],
258        bounds: ComputeBudgetBounds,
259    ) -> Result<SubmitSignedSponsoredTxData, VaultTxError> {
260        let transaction = verify_vault_open(to_sign, intent, tables, bounds)?;
261        ensure_system_time_ahead("signature_deadline", to_sign.signature_deadline)?;
262        Ok(SubmitSignedSponsoredTxData {
263            request_id: Uuid::new_v4(),
264            order_id: to_sign.order_id,
265            tx_base64: transaction.sign(self.operator.as_ref())?,
266        })
267    }
268
269    pub async fn submit_signed_sponsored_tx(
270        &self,
271        payload: SubmitSignedSponsoredTxData,
272    ) -> Result<(), VaultRequestError> {
273        self.command(ClientMessage::SubmitSignedSponsoredTx(payload))
274            .await
275    }
276
277    pub async fn swap_quote(
278        &self,
279        payload: SwapQuoteData,
280    ) -> Result<SwapQuoteResultData, VaultRequestError> {
281        self.request(
282            ClientMessage::SwapQuote(payload),
283            "swap_quote_result",
284            |message| match message {
285                ServerMessage::SwapQuoteResult(data) => Some(data.clone()),
286                _ => None,
287            },
288        )
289        .await
290    }
291
292    /// Signs only if the served swap carries exactly the requested and quoted terms within `bounds`.
293    pub fn verify_and_sign_swap(
294        &self,
295        intent: &VaultSwapIntent<'_>,
296        tables: &[AddressLookupTableAccount],
297        bounds: ComputeBudgetBounds,
298    ) -> Result<SwapSubmitData, VaultTxError> {
299        let transaction = verify_vault_swap(intent, tables, bounds)?;
300        ensure_deadline_ahead("valid_until", &intent.quote.valid_until)?;
301        Ok(SwapSubmitData {
302            request_id: Uuid::new_v4(),
303            execution_id: intent.quote.execution_id,
304            signed_tx_base64: transaction.sign(self.operator.as_ref())?,
305        })
306    }
307
308    /// Submits only in `connection_epoch` and only if `gate` admits it; resolves on
309    /// the durable enqueue receipt. [`VaultRequestError::delivery`] classifies failures.
310    pub async fn swap_submit_in_epoch(
311        &self,
312        payload: SwapSubmitData,
313        connection_epoch: u64,
314        gate: &impl SendGate,
315    ) -> Result<SwapSubmitResultData, VaultRequestError> {
316        self.gated_request(
317            gate,
318            connection_epoch,
319            ClientMessage::SwapSubmit(payload),
320            "swap_submit_result",
321            |message| match message {
322                ServerMessage::SwapSubmitResult(data) => Some(data.clone()),
323                _ => None,
324            },
325        )
326        .await
327    }
328
329    /// Terminal outcomes also arrive as `SwapSettled` pushes.
330    pub async fn swap_status(
331        &self,
332        execution_id: OrderId,
333    ) -> Result<SwapStatusMessage, VaultRequestError> {
334        self.request(
335            ClientMessage::GetSwapStatus(GetSwapStatusMessage {
336                request_id: Uuid::new_v4(),
337                execution_id,
338            }),
339            "swap_status",
340            |message| match message {
341                ServerMessage::SwapStatus(data) => Some(data.clone()),
342                _ => None,
343            },
344        )
345        .await
346    }
347
348    pub async fn subscribe(&self, payload: SubscribeData) -> Result<(), ManagedWsError> {
349        self.send(ClientMessage::Subscribe(payload)).await
350    }
351
352    pub async fn unsubscribe(&self, payload: UnsubscribeData) -> Result<(), ManagedWsError> {
353        self.send(ClientMessage::Unsubscribe(payload)).await
354    }
355
356    pub fn subscribe_messages(&self) -> ManagedMessageReceiver {
357        self.inner.subscribe_messages()
358    }
359
360    pub fn subscribe_reference_changes(&self) -> ReferenceChangeReceiver {
361        self.inner.subscribe_reference_changes()
362    }
363
364    pub fn subscribe_projections(&self) -> VaultProjectionReceiver {
365        VaultProjectionReceiver::new(self.inner.subscribe_messages())
366    }
367
368    pub fn subscribe_events(&self) -> broadcast::Receiver<ManagedWsEvent> {
369        self.inner.subscribe_events()
370    }
371
372    #[must_use]
373    pub fn state(&self) -> ManagedWsState {
374        self.inner.state()
375    }
376
377    pub async fn wait_until_ready(&self) -> Result<u64, WaitUntilReadyError> {
378        self.inner.wait_until_ready().await
379    }
380
381    #[must_use]
382    pub const fn raw(&self) -> &ManagedWsHandle {
383        &self.inner
384    }
385
386    #[must_use]
387    pub fn into_raw(self) -> ManagedWsHandle {
388        self.inner
389    }
390
391    pub async fn close(&self) -> Result<(), ManagedWsError> {
392        self.inner.close().await
393    }
394
395    async fn send(&self, message: ClientMessage) -> Result<(), ManagedWsError> {
396        self.inner.ensure_ready()?;
397        self.inner.send(message).await
398    }
399
400    async fn command(&self, message: ClientMessage) -> Result<(), VaultRequestError> {
401        self.request(message, "command_accepted", |message| match message {
402            ServerMessage::CommandAccepted(_) => Some(()),
403            _ => None,
404        })
405        .await
406    }
407
408    #[cfg(feature = "chain-rpc")]
409    pub(super) async fn gated_command(
410        &self,
411        gate: &impl SendGate,
412        connection_epoch: u64,
413        message: ClientMessage,
414    ) -> Result<(), VaultRequestError> {
415        self.gated_request(
416            gate,
417            connection_epoch,
418            message,
419            "command_accepted",
420            |message| match message {
421                ServerMessage::CommandAccepted(_) => Some(()),
422                _ => None,
423            },
424        )
425        .await
426    }
427
428    async fn gated_request<T>(
429        &self,
430        gate: &impl SendGate,
431        connection_epoch: u64,
432        message: ClientMessage,
433        expected: &'static str,
434        extract: fn(&ServerMessage) -> Option<T>,
435    ) -> Result<T, VaultRequestError> {
436        let response = gate
437            .admit(connection_epoch, || {
438                self.inner
439                    .try_send_await_in_epoch(message, self.request_timeout, connection_epoch)
440            })
441            .ok_or(VaultRequestError::Gated)??
442            .await?;
443        Self::correlated(response, expected, extract)
444    }
445
446    async fn request<T>(
447        &self,
448        message: ClientMessage,
449        expected: &'static str,
450        extract: fn(&ServerMessage) -> Option<T>,
451    ) -> Result<T, VaultRequestError> {
452        self.inner.ensure_ready()?;
453        let response = self.inner.send_await(message, self.request_timeout).await?;
454        Self::correlated(response, expected, extract)
455    }
456
457    fn correlated<T>(
458        response: Arc<ServerMessage>,
459        expected: &'static str,
460        extract: fn(&ServerMessage) -> Option<T>,
461    ) -> Result<T, VaultRequestError> {
462        if let ServerMessage::RequestError(error) = &*response {
463            return Err(VaultRequestError::Rejected {
464                request_id: error.request_id,
465                response,
466            });
467        }
468        extract(&response).ok_or(VaultRequestError::UnexpectedResponse { expected, response })
469    }
470}
acta_maker_sdk/ws/vault/client.rs

acta_maker_sdk/ws/vault/
client.rs
