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

acta_maker_sdk/ws/vault/
client.rs
