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::wire::{WireError, decode_base58_32};
11use crate::ws::managed::{
12    ManagedMessageReceiver, ManagedWsConfigError, ManagedWsError, ManagedWsEvent, ManagedWsHandle,
13    ManagedWsState, ReferenceChangeReceiver, SendAwaitError, SendTicket, WaitUntilReadyError,
14    spawn_managed_ws,
15};
16use crate::ws::types::{
17    AcceptQuoteMessage, BatchQuotesMessage, CancelAllQuotesMessage, CancelQuoteData, CancelRfqData,
18    ClientMessage, QuoteMessage, ReplaceQuoteMessage, RfqRequestMessage, ServerError,
19    ServerMessage, SponsoredTxToSignData, SubmitSignedSponsoredTxData, SubscribeData,
20    SwapQuoteData, SwapQuoteResultData, SwapSubmitData, SwapSubmitResultData, UnsubscribeData,
21    VaultAuthSuccessData,
22};
23
24use solana_sdk::message::AddressLookupTableAccount;
25
26use crate::chain::vault::{
27    ComputeBudgetBounds, VaultOpenIntent, VaultSwapIntent, verify_vault_open, verify_vault_swap,
28};
29
30use super::config::VaultOperatorConfig;
31use super::projection::VaultProjectionReceiver;
32use super::tx::{VaultTxError, ensure_deadline_ahead, ensure_system_time_ahead};
33
34#[derive(Debug, thiserror::Error)]
35#[non_exhaustive]
36pub enum VaultClientError {
37    #[error("vault_pda is not a base58 32-byte address")]
38    InvalidVaultPda(#[source] WireError),
39    #[error(transparent)]
40    InvalidConfig(#[from] ManagedWsConfigError),
41}
42
43#[derive(Debug, thiserror::Error)]
44#[non_exhaustive]
45pub enum VaultRequestError {
46    #[error(transparent)]
47    Session(#[from] ManagedWsError),
48    #[error(transparent)]
49    Transport(#[from] SendAwaitError),
50    #[error("server rejected request {request_id}")]
51    Rejected {
52        request_id: Uuid,
53        response: Arc<ServerMessage>,
54    },
55    #[error("expected {expected}, received a different correlated response")]
56    UnexpectedResponse {
57        expected: &'static str,
58        response: Arc<ServerMessage>,
59    },
60}
61
62impl VaultRequestError {
63    #[must_use]
64    pub fn server_error(&self) -> Option<&ServerError> {
65        match self {
66            Self::Rejected { response, .. } => match &**response {
67                ServerMessage::RequestError(error) => Some(&error.error),
68                _ => None,
69            },
70            _ => None,
71        }
72    }
73}
74
75/// Vault operator session on `/vault`; every reconnect re-runs `StartVaultAuth`.
76#[derive(Clone)]
77pub struct VaultOperatorClient {
78    inner: ManagedWsHandle,
79    vault_pda: String,
80    vault_owner: [u8; 32],
81    operator: Arc<dyn SignerLike + Send + Sync>,
82    request_timeout: Duration,
83}
84
85impl VaultOperatorClient {
86    pub fn spawn(config: VaultOperatorConfig) -> Result<Self, VaultClientError> {
87        let vault_owner =
88            decode_base58_32(&config.vault_pda).map_err(VaultClientError::InvalidVaultPda)?;
89        Ok(Self {
90            inner: spawn_managed_ws(config.managed)?,
91            vault_pda: config.vault_pda,
92            vault_owner,
93            operator: config.operator,
94            request_timeout: config.request_timeout,
95        })
96    }
97
98    #[must_use]
99    pub fn vault_pda(&self) -> &str {
100        &self.vault_pda
101    }
102
103    /// Vault address used as `maker` in the order preimage.
104    #[must_use]
105    pub const fn vault_owner(&self) -> [u8; 32] {
106        self.vault_owner
107    }
108
109    #[must_use]
110    pub fn operator_pubkey(&self) -> String {
111        self.operator.pubkey_base58()
112    }
113
114    /// `None` before authentication and after it ends.
115    #[must_use]
116    pub fn auth(&self) -> Option<VaultAuthSuccessData> {
117        self.inner.vault_auth()
118    }
119
120    #[must_use]
121    pub fn permissions(&self) -> Option<VaultDelegatePermissions> {
122        self.auth().map(|auth| auth.permissions)
123    }
124
125    #[must_use]
126    pub fn can_maker(&self) -> bool {
127        self.inner.is_ready()
128            && self.auth().is_some_and(|auth| {
129                matches!(
130                    auth.operator_kind,
131                    crate::types::VaultOperatorKind::Delegate
132                        | crate::types::VaultOperatorKind::Governance
133                ) && auth.permissions.can_maker()
134            })
135    }
136
137    #[must_use]
138    pub fn can_taker(&self) -> bool {
139        self.inner.is_ready()
140            && self
141                .permissions()
142                .is_some_and(VaultDelegatePermissions::can_taker)
143    }
144
145    #[must_use]
146    pub fn subscribe_vault_auth(&self) -> watch::Receiver<Option<VaultAuthSuccessData>> {
147        self.inner.subscribe_vault_auth()
148    }
149
150    pub fn sign_quote(&self, quote: &QuoteBuilder<'_>) -> Result<QuoteMessage, QuoteBuildError> {
151        quote
152            .clone()
153            .maker_owner(self.vault_owner)
154            .sign(self.operator.as_ref())
155    }
156
157    pub async fn quote(&self, payload: QuoteMessage) -> Result<(), ManagedWsError> {
158        self.send(ClientMessage::Quote(payload)).await
159    }
160
161    pub fn try_quote(&self, payload: QuoteMessage) -> Result<SendTicket, ManagedWsError> {
162        self.inner.ensure_ready()?;
163        self.inner.try_send(ClientMessage::Quote(payload))
164    }
165
166    pub async fn replace_quote(&self, payload: ReplaceQuoteMessage) -> Result<(), ManagedWsError> {
167        self.send(ClientMessage::ReplaceQuote(payload)).await
168    }
169
170    pub async fn batch_quotes(&self, payload: BatchQuotesMessage) -> Result<(), ManagedWsError> {
171        self.send(ClientMessage::BatchQuotes(payload)).await
172    }
173
174    pub async fn cancel_quote(&self, payload: CancelQuoteData) -> Result<(), ManagedWsError> {
175        self.send(ClientMessage::CancelQuote(payload)).await
176    }
177
178    pub async fn cancel_all_quotes(
179        &self,
180        payload: CancelAllQuotesMessage,
181    ) -> Result<(), ManagedWsError> {
182        self.send(ClientMessage::CancelAllQuotes(payload)).await
183    }
184
185    pub async fn rfq_request(&self, payload: RfqRequestMessage) -> Result<(), ManagedWsError> {
186        self.send(ClientMessage::RfqRequest(payload)).await
187    }
188
189    pub async fn cancel_rfq(&self, payload: CancelRfqData) -> Result<(), ManagedWsError> {
190        self.send(ClientMessage::CancelRfq(payload)).await
191    }
192
193    pub async fn accept_quote(&self, payload: AcceptQuoteMessage) -> Result<(), ManagedWsError> {
194        self.send(ClientMessage::AcceptQuote(payload)).await
195    }
196
197    /// Signs only if the served transaction opens exactly `intent` within `bounds`.
198    pub fn verify_and_sign_sponsored_tx(
199        &self,
200        to_sign: &SponsoredTxToSignData,
201        intent: &VaultOpenIntent<'_>,
202        tables: &[AddressLookupTableAccount],
203        bounds: ComputeBudgetBounds,
204    ) -> Result<SubmitSignedSponsoredTxData, VaultTxError> {
205        let transaction = verify_vault_open(to_sign, intent, tables, bounds)?;
206        ensure_system_time_ahead("signature_deadline", to_sign.signature_deadline)?;
207        Ok(SubmitSignedSponsoredTxData {
208            order_id: to_sign.order_id,
209            tx_base64: transaction.sign(self.operator.as_ref())?,
210        })
211    }
212
213    pub async fn submit_signed_sponsored_tx(
214        &self,
215        payload: SubmitSignedSponsoredTxData,
216    ) -> Result<(), ManagedWsError> {
217        self.send(ClientMessage::SubmitSignedSponsoredTx(payload))
218            .await
219    }
220
221    pub async fn swap_quote(
222        &self,
223        payload: SwapQuoteData,
224    ) -> Result<SwapQuoteResultData, VaultRequestError> {
225        self.request(
226            ClientMessage::SwapQuote(payload),
227            "swap_quote_result",
228            |message| match message {
229                ServerMessage::SwapQuoteResult(data) => Ok(data),
230                other => Err(other),
231            },
232        )
233        .await
234    }
235
236    /// Signs only if the served swap carries exactly the requested and quoted terms within `bounds`.
237    pub fn verify_and_sign_swap(
238        &self,
239        intent: &VaultSwapIntent<'_>,
240        tables: &[AddressLookupTableAccount],
241        bounds: ComputeBudgetBounds,
242    ) -> Result<SwapSubmitData, VaultTxError> {
243        let transaction = verify_vault_swap(intent, tables, bounds)?;
244        ensure_deadline_ahead("valid_until", &intent.quote.valid_until)?;
245        Ok(SwapSubmitData {
246            request_id: Uuid::new_v4(),
247            execution_id: intent.quote.execution_id,
248            signed_tx_base64: transaction.sign(self.operator.as_ref())?,
249        })
250    }
251
252    pub async fn swap_submit(
253        &self,
254        payload: SwapSubmitData,
255    ) -> Result<SwapSubmitResultData, VaultRequestError> {
256        self.request(
257            ClientMessage::SwapSubmit(payload),
258            "swap_submit_result",
259            |message| match message {
260                ServerMessage::SwapSubmitResult(data) => Ok(data),
261                other => Err(other),
262            },
263        )
264        .await
265    }
266
267    pub async fn subscribe(&self, payload: SubscribeData) -> Result<(), ManagedWsError> {
268        self.send(ClientMessage::Subscribe(payload)).await
269    }
270
271    pub async fn unsubscribe(&self, payload: UnsubscribeData) -> Result<(), ManagedWsError> {
272        self.send(ClientMessage::Unsubscribe(payload)).await
273    }
274
275    pub fn subscribe_messages(&self) -> ManagedMessageReceiver {
276        self.inner.subscribe_messages()
277    }
278
279    pub fn subscribe_reference_changes(&self) -> ReferenceChangeReceiver {
280        self.inner.subscribe_reference_changes()
281    }
282
283    pub fn subscribe_projections(&self) -> VaultProjectionReceiver {
284        VaultProjectionReceiver::new(self.inner.subscribe_messages())
285    }
286
287    pub fn subscribe_events(&self) -> broadcast::Receiver<ManagedWsEvent> {
288        self.inner.subscribe_events()
289    }
290
291    #[must_use]
292    pub fn state(&self) -> ManagedWsState {
293        self.inner.state()
294    }
295
296    pub async fn wait_until_ready(&self) -> Result<u64, WaitUntilReadyError> {
297        self.inner.wait_until_ready().await
298    }
299
300    #[must_use]
301    pub const fn raw(&self) -> &ManagedWsHandle {
302        &self.inner
303    }
304
305    #[must_use]
306    pub fn into_raw(self) -> ManagedWsHandle {
307        self.inner
308    }
309
310    pub async fn close(&self) -> Result<(), ManagedWsError> {
311        self.inner.close().await
312    }
313
314    async fn send(&self, message: ClientMessage) -> Result<(), ManagedWsError> {
315        self.inner.ensure_ready()?;
316        self.inner.send(message).await
317    }
318
319    async fn request<T>(
320        &self,
321        message: ClientMessage,
322        expected: &'static str,
323        extract: fn(ServerMessage) -> Result<T, ServerMessage>,
324    ) -> Result<T, VaultRequestError> {
325        self.inner.ensure_ready()?;
326        let response = self.inner.send_await(message, self.request_timeout).await?;
327        if let ServerMessage::RequestError(error) = &*response {
328            return Err(VaultRequestError::Rejected {
329                request_id: error.request_id,
330                response,
331            });
332        }
333        let owned = match Arc::try_unwrap(response) {
334            Ok(owned) => owned,
335            Err(shared) => (*shared).clone(),
336        };
337        extract(owned).map_err(|other| VaultRequestError::UnexpectedResponse {
338            expected,
339            response: Arc::new(other),
340        })
341    }
342}
acta_maker_sdk/ws/vault/client.rs

acta_maker_sdk/ws/vault/
client.rs
