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    /// Ready to quote buy RFQs, the vault writing what the wallet buys.
163    #[must_use]
164    pub fn can_write_buys(&self) -> bool {
165        self.inner.is_ready()
166            && self
167                .permissions()
168                .is_some_and(VaultDelegatePermissions::can_write_buys)
169    }
170
171    #[must_use]
172    pub fn can_taker(&self) -> bool {
173        self.inner.is_ready()
174            && self
175                .permissions()
176                .is_some_and(VaultDelegatePermissions::can_taker)
177    }
178
179    #[must_use]
180    pub fn subscribe_vault_auth(&self) -> watch::Receiver<Option<VaultAuthSuccessData>> {
181        self.inner.subscribe_vault_auth()
182    }
183
184    pub fn sign_quote(&self, quote: &QuoteBuilder<'_>) -> Result<QuoteMessage, QuoteBuildError> {
185        quote
186            .clone()
187            .maker_owner(self.vault_pda)
188            .sign(self.operator.as_ref())
189    }
190
191    /// Send only in `connection_epoch`, the epoch whose recovery the strategy
192    /// applied; a reconnect in between rejects the quote with `NotReady`.
193    pub async fn quote_in_epoch(
194        &self,
195        payload: QuoteMessage,
196        connection_epoch: u64,
197    ) -> Result<(), ManagedWsError> {
198        self.inner
199            .send_in_epoch(ClientMessage::Quote(payload), connection_epoch)
200            .await
201    }
202
203    pub fn try_quote_in_epoch(
204        &self,
205        payload: QuoteMessage,
206        connection_epoch: u64,
207    ) -> Result<SendTicket, ManagedWsError> {
208        self.inner
209            .try_send_in_epoch(ClientMessage::Quote(payload), connection_epoch)
210    }
211
212    pub async fn replace_in_epoch(
213        &self,
214        payload: ReplaceQuoteMessage,
215        connection_epoch: u64,
216    ) -> Result<(), ManagedWsError> {
217        self.inner
218            .send_in_epoch(ClientMessage::ReplaceQuote(payload), connection_epoch)
219            .await
220    }
221
222    pub async fn batch_quotes_in_epoch(
223        &self,
224        payload: BatchQuotesMessage,
225        connection_epoch: u64,
226    ) -> Result<(), ManagedWsError> {
227        self.inner
228            .send_in_epoch(ClientMessage::BatchQuotes(payload), connection_epoch)
229            .await
230    }
231
232    pub async fn cancel_quote(&self, payload: CancelQuoteData) -> Result<(), ManagedWsError> {
233        self.send(ClientMessage::CancelQuote(payload)).await
234    }
235
236    pub async fn cancel_all_quotes(
237        &self,
238        payload: CancelAllQuotesMessage,
239    ) -> Result<(), ManagedWsError> {
240        self.send(ClientMessage::CancelAllQuotes(payload)).await
241    }
242
243    /// Resolves on the server's receipt; `RfqCreated` still arrives as a push.
244    pub async fn rfq_request(&self, payload: RfqRequestMessage) -> Result<(), VaultRequestError> {
245        if payload.side.is_taker_buy() {
246            return Err(VaultRequestError::TakerBuyUnsupported);
247        }
248        self.command(ClientMessage::RfqRequest(payload)).await
249    }
250
251    pub async fn cancel_rfq(&self, payload: CancelRfqData) -> Result<(), ManagedWsError> {
252        self.send(ClientMessage::CancelRfq(payload)).await
253    }
254
255    pub async fn accept_quote(&self, payload: AcceptQuoteMessage) -> Result<(), VaultRequestError> {
256        self.command(ClientMessage::AcceptQuote(payload)).await
257    }
258
259    /// Signs only if the served transaction opens exactly `intent` within `bounds`.
260    pub fn verify_and_sign_sponsored_tx(
261        &self,
262        to_sign: &SponsoredTxToSignData,
263        intent: &VaultOpenIntent<'_>,
264        tables: &[AddressLookupTableAccount],
265        bounds: ComputeBudgetBounds,
266    ) -> Result<SubmitSignedSponsoredTxData, VaultTxError> {
267        self.verify_and_sign_sponsored_tx_with_participant_signature(
268            to_sign, intent, tables, bounds,
269        )
270        .map(|(submit, _)| submit)
271    }
272
273    pub(crate) fn verify_and_sign_sponsored_tx_with_participant_signature(
274        &self,
275        to_sign: &SponsoredTxToSignData,
276        intent: &VaultOpenIntent<'_>,
277        tables: &[AddressLookupTableAccount],
278        bounds: ComputeBudgetBounds,
279    ) -> Result<(SubmitSignedSponsoredTxData, String), VaultTxError> {
280        let expected_participant = match intent {
281            VaultOpenIntent::Sale { writer, .. }
282            | VaultOpenIntent::BetweenVaults { writer, .. } => writer.actor(),
283            VaultOpenIntent::Fill { taker, .. } => *taker,
284            VaultOpenIntent::Buy { holder, .. } => *holder,
285        };
286        let signer = Pubkey::new_from_array(self.operator.pubkey_bytes());
287        if signer != expected_participant {
288            return Err(VaultIntentError::ParticipantSigner {
289                expected: expected_participant,
290                found: signer,
291            }
292            .into());
293        }
294        let transaction = verify_vault_open(to_sign, intent, tables, bounds)?;
295        ensure_system_time_ahead("signature_deadline", to_sign.signature_deadline)?;
296        let (tx_base64, participant_signature) =
297            transaction.sign_with_participant_signature(self.operator.as_ref())?;
298        Ok((
299            SubmitSignedSponsoredTxData {
300                request_id: Uuid::new_v4(),
301                order_id: to_sign.order_id,
302                tx_base64,
303            },
304            participant_signature,
305        ))
306    }
307
308    pub async fn submit_signed_sponsored_tx(
309        &self,
310        payload: SubmitSignedSponsoredTxData,
311    ) -> Result<(), VaultRequestError> {
312        self.command(ClientMessage::SubmitSignedSponsoredTx(payload))
313            .await
314    }
315
316    pub async fn swap_quote(
317        &self,
318        payload: SwapQuoteData,
319    ) -> Result<SwapQuoteResultData, VaultRequestError> {
320        self.request(
321            ClientMessage::SwapQuote(payload),
322            "swap_quote_result",
323            |message| match message {
324                ServerMessage::SwapQuoteResult(data) => Some(data.clone()),
325                _ => None,
326            },
327        )
328        .await
329    }
330
331    /// Signs only if the served swap carries exactly the requested and quoted terms within `bounds`.
332    pub fn verify_and_sign_swap(
333        &self,
334        intent: &VaultSwapIntent<'_>,
335        tables: &[AddressLookupTableAccount],
336        bounds: ComputeBudgetBounds,
337    ) -> Result<SwapSubmitData, VaultTxError> {
338        let transaction = verify_vault_swap(intent, tables, bounds)?;
339        ensure_deadline_ahead("valid_until", &intent.quote.valid_until)?;
340        Ok(SwapSubmitData {
341            request_id: Uuid::new_v4(),
342            execution_id: intent.quote.execution_id,
343            signed_tx_base64: transaction.sign(self.operator.as_ref())?,
344        })
345    }
346
347    /// Submits only in `connection_epoch` and only if `gate` admits it; resolves on
348    /// the durable enqueue receipt. [`VaultRequestError::delivery`] classifies failures.
349    pub async fn swap_submit_in_epoch(
350        &self,
351        payload: SwapSubmitData,
352        connection_epoch: u64,
353        gate: &impl SendGate,
354    ) -> Result<SwapSubmitResultData, VaultRequestError> {
355        self.gated_request(
356            gate,
357            connection_epoch,
358            ClientMessage::SwapSubmit(payload),
359            "swap_submit_result",
360            |message| match message {
361                ServerMessage::SwapSubmitResult(data) => Some(data.clone()),
362                _ => None,
363            },
364        )
365        .await
366    }
367
368    /// Terminal outcomes also arrive as `SwapSettled` pushes.
369    pub async fn swap_status(
370        &self,
371        execution_id: OrderId,
372    ) -> Result<SwapStatusMessage, VaultRequestError> {
373        self.request(
374            ClientMessage::GetSwapStatus(GetSwapStatusMessage {
375                request_id: Uuid::new_v4(),
376                execution_id,
377            }),
378            "swap_status",
379            |message| match message {
380                ServerMessage::SwapStatus(data) => Some(data.clone()),
381                _ => None,
382            },
383        )
384        .await
385    }
386
387    pub async fn subscribe(&self, payload: SubscribeData) -> Result<(), ManagedWsError> {
388        self.send(ClientMessage::Subscribe(payload)).await
389    }
390
391    pub async fn unsubscribe(&self, payload: UnsubscribeData) -> Result<(), ManagedWsError> {
392        self.send(ClientMessage::Unsubscribe(payload)).await
393    }
394
395    pub fn subscribe_messages(&self) -> ManagedMessageReceiver {
396        self.inner.subscribe_messages()
397    }
398
399    pub fn subscribe_reference_changes(&self) -> ReferenceChangeReceiver {
400        self.inner.subscribe_reference_changes()
401    }
402
403    pub fn subscribe_projections(&self) -> VaultProjectionReceiver {
404        VaultProjectionReceiver::new(self.inner.subscribe_messages())
405    }
406
407    pub fn subscribe_events(&self) -> broadcast::Receiver<ManagedWsEvent> {
408        self.inner.subscribe_events()
409    }
410
411    #[must_use]
412    pub fn state(&self) -> ManagedWsState {
413        self.inner.state()
414    }
415
416    pub async fn wait_until_ready(&self) -> Result<u64, WaitUntilReadyError> {
417        self.inner.wait_until_ready().await
418    }
419
420    #[must_use]
421    pub const fn raw(&self) -> &ManagedWsHandle {
422        &self.inner
423    }
424
425    #[must_use]
426    pub fn into_raw(self) -> ManagedWsHandle {
427        self.inner
428    }
429
430    pub async fn close(&self) -> Result<(), ManagedWsError> {
431        self.inner.close().await
432    }
433
434    async fn send(&self, message: ClientMessage) -> Result<(), ManagedWsError> {
435        self.inner.ensure_ready()?;
436        self.inner.send(message).await
437    }
438
439    async fn command(&self, message: ClientMessage) -> Result<(), VaultRequestError> {
440        self.request(message, "command_accepted", |message| match message {
441            ServerMessage::CommandAccepted(_) => Some(()),
442            _ => None,
443        })
444        .await
445    }
446
447    #[cfg(feature = "chain-rpc")]
448    pub(super) async fn gated_command(
449        &self,
450        gate: &impl SendGate,
451        connection_epoch: u64,
452        message: ClientMessage,
453    ) -> Result<(), VaultRequestError> {
454        self.gated_request(
455            gate,
456            connection_epoch,
457            message,
458            "command_accepted",
459            |message| match message {
460                ServerMessage::CommandAccepted(_) => Some(()),
461                _ => None,
462            },
463        )
464        .await
465    }
466
467    async fn gated_request<T>(
468        &self,
469        gate: &impl SendGate,
470        connection_epoch: u64,
471        message: ClientMessage,
472        expected: &'static str,
473        extract: fn(&ServerMessage) -> Option<T>,
474    ) -> Result<T, VaultRequestError> {
475        let response = gate
476            .admit(connection_epoch, || {
477                self.inner
478                    .try_send_await_in_epoch(message, self.request_timeout, connection_epoch)
479            })
480            .ok_or(VaultRequestError::Gated)??
481            .await?;
482        Self::correlated(response, expected, extract)
483    }
484
485    async fn request<T>(
486        &self,
487        message: ClientMessage,
488        expected: &'static str,
489        extract: fn(&ServerMessage) -> Option<T>,
490    ) -> Result<T, VaultRequestError> {
491        self.inner.ensure_ready()?;
492        let response = self.inner.send_await(message, self.request_timeout).await?;
493        Self::correlated(response, expected, extract)
494    }
495
496    fn correlated<T>(
497        response: Arc<ServerMessage>,
498        expected: &'static str,
499        extract: fn(&ServerMessage) -> Option<T>,
500    ) -> Result<T, VaultRequestError> {
501        if let ServerMessage::RequestError(error) = &*response {
502            return Err(VaultRequestError::Rejected {
503                request_id: error.request_id,
504                response,
505            });
506        }
507        extract(&response).ok_or(VaultRequestError::UnexpectedResponse { expected, response })
508    }
509}
acta_maker_sdk/ws/vault/client.rs

acta_maker_sdk/ws/vault/
client.rs
