strata_sdk/
maker_stream.rs

1use std::time::Duration;
2
3use futures_util::{SinkExt, StreamExt};
4use tokio::net::TcpStream;
5use tokio_tungstenite::tungstenite::Message;
6use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
7
8use super::account_stream::validate_account_state;
9use super::market_stream::platform_websocket_url;
10use super::*;
11
12pub const MAKER_STREAM_AUTH_DOMAIN: &str = "strata:mm-fills-stream:v2";
13const AUTH_TIMEOUT: Duration = Duration::from_secs(10);
14
15type PlatformSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
16
17/// One externally authenticated owner-only maker stream. The SDK retains no
18/// signer and fails closed on any identity or sequence mismatch.
19pub struct MakerStream {
20    market_id: String,
21    wallet_address: String,
22    stream_id: String,
23    sequence: u64,
24    initial_snapshot: Option<PlatformMakerEvent>,
25    socket: PlatformSocket,
26}
27
28impl std::fmt::Debug for MakerStream {
29    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30        formatter
31            .debug_struct("MakerStream")
32            .field("market_id", &self.market_id)
33            .field("wallet_address", &self.wallet_address)
34            .field("stream_id", &self.stream_id)
35            .field("sequence", &self.sequence)
36            .finish_non_exhaustive()
37    }
38}
39
40impl MakerStream {
41    /// `signer` answers the server's compatibility challenge with a signature;
42    /// `None` opens the public stream by wallet address (`{"type":"open"}`).
43    pub(crate) async fn connect<S: AccountSigner + ?Sized>(
44        client: &StrataClient,
45        market_id: &str,
46        wallet_address: &str,
47        signer: Option<&S>,
48    ) -> Result<Self, SdkError> {
49        let market_id = validate_platform_market_id(market_id)?;
50        let wallet_address = canonical_public_key(wallet_address, "maker wallet address")?;
51        let url = platform_websocket_url(
52            &client.base_url,
53            &format!("v2/markets/{market_id}/makers/{wallet_address}/stream"),
54        )?;
55        let (mut socket, _) = tokio_tungstenite::connect_async(url.as_str())
56            .await
57            .map_err(|error| SdkError::Stream(error.to_string()))?;
58
59        let challenge_frame = next_text(&mut socket, "maker authentication challenge").await?;
60        let challenge_event: PlatformMakerEvent = serde_json::from_str(&challenge_frame)
61            .map_err(|error| SdkError::InvalidResponse(error.to_string()))?;
62        let challenge = match challenge_event {
63            PlatformMakerEvent::AuthChallenge {
64                schema_version,
65                contract_version,
66                market_id: response_market,
67                wallet_address: response_wallet,
68                challenge,
69                server_time_ms,
70                expires_at_ms,
71            } => {
72                validate_platform_version(schema_version, &contract_version)?;
73                if response_market != market_id
74                    || response_wallet != wallet_address
75                    || expires_at_ms <= server_time_ms
76                    || challenge.len() != 64
77                    || !challenge
78                        .bytes()
79                        .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
80                {
81                    return Err(SdkError::InvalidResponse(
82                        "maker stream authentication bindings are invalid".to_owned(),
83                    ));
84                }
85                challenge
86            }
87            _ => {
88                return Err(SdkError::InvalidResponse(
89                    "maker stream did not begin with authentication".to_owned(),
90                ))
91            }
92        };
93
94        let answer = match signer {
95            Some(signer) => {
96                let message = maker_stream_auth_message(&market_id, &wallet_address, &challenge)?;
97                let signature = signer
98                    .sign_message(&message)
99                    .await
100                    .map_err(SdkError::Signer)?;
101                if signature.len() != 64 {
102                    return Err(SdkError::Signer(
103                        "maker signer must return a 64-byte Ed25519 signature".to_owned(),
104                    ));
105                }
106                serde_json::json!({
107                    "type": "authenticate",
108                    "signature": hex::encode(signature),
109                })
110            }
111            // Public read: no signature needed.
112            None => serde_json::json!({ "type": "open" }),
113        };
114        socket
115            .send(Message::Text(answer.to_string().into()))
116            .await
117            .map_err(|error| SdkError::Stream(error.to_string()))?;
118
119        let snapshot_frame = next_text(&mut socket, "signed maker snapshot").await?;
120        let snapshot: PlatformMakerEvent = serde_json::from_str(&snapshot_frame)
121            .map_err(|error| SdkError::InvalidResponse(error.to_string()))?;
122        let (stream_id, sequence) = match &snapshot {
123            PlatformMakerEvent::MakerSnapshot {
124                schema_version,
125                contract_version,
126                market_id: response_market,
127                wallet_address: response_wallet,
128                stream_id,
129                sequence,
130                status,
131                fills,
132                ..
133            } => {
134                validate_maker_identity(
135                    *schema_version,
136                    contract_version,
137                    response_market,
138                    response_wallet,
139                    &market_id,
140                    &wallet_address,
141                )?;
142                if !valid_handle(stream_id, "maker_stream_") {
143                    return Err(SdkError::InvalidResponse(
144                        "maker stream identity is invalid".to_owned(),
145                    ));
146                }
147                validate_maker_stream_state(status, fills, &market_id, &wallet_address)?;
148                (
149                    stream_id.clone(),
150                    validate_response_atoms(sequence, "sequence", false)?,
151                )
152            }
153            _ => {
154                return Err(SdkError::InvalidResponse(
155                    "maker authentication did not return a signed snapshot".to_owned(),
156                ))
157            }
158        };
159
160        Ok(Self {
161            market_id,
162            wallet_address,
163            stream_id,
164            sequence,
165            initial_snapshot: Some(snapshot),
166            socket,
167        })
168    }
169
170    pub fn market_id(&self) -> &str {
171        &self.market_id
172    }
173
174    pub fn wallet_address(&self) -> &str {
175        &self.wallet_address
176    }
177
178    /// The first call returns the authenticated snapshot received during
179    /// connection. Later calls return sequenced fill, status, heartbeat, or
180    /// recovery-snapshot events.
181    pub async fn next_event(&mut self) -> Result<Option<PlatformMakerEvent>, SdkError> {
182        if let Some(snapshot) = self.initial_snapshot.take() {
183            return Ok(Some(snapshot));
184        }
185        loop {
186            let Some(frame) = self.socket.next().await else {
187                return Ok(None);
188            };
189            let frame = frame.map_err(|error| SdkError::Stream(error.to_string()))?;
190            match frame {
191                Message::Text(text) => {
192                    let event: PlatformMakerEvent = serde_json::from_str(&text)
193                        .map_err(|error| SdkError::InvalidResponse(error.to_string()))?;
194                    if let Err(error) = self.validate_event(&event) {
195                        let _ = self.socket.close(None).await;
196                        return Err(error);
197                    }
198                    return Ok(Some(event));
199                }
200                Message::Ping(payload) => {
201                    self.socket
202                        .send(Message::Pong(payload))
203                        .await
204                        .map_err(|error| SdkError::Stream(error.to_string()))?;
205                }
206                Message::Pong(_) => {}
207                Message::Close(_) => return Ok(None),
208                _ => {
209                    let _ = self.socket.close(None).await;
210                    return Err(SdkError::InvalidResponse(
211                        "maker stream sent a non-text data frame".to_owned(),
212                    ));
213                }
214            }
215        }
216    }
217
218    pub async fn close(&mut self) -> Result<(), SdkError> {
219        self.socket
220            .close(None)
221            .await
222            .map_err(|error| SdkError::Stream(error.to_string()))
223    }
224
225    fn validate_event(&mut self, event: &PlatformMakerEvent) -> Result<(), SdkError> {
226        match event {
227            PlatformMakerEvent::AuthChallenge { .. } => {
228                return Err(SdkError::InvalidResponse(
229                    "maker stream challenged after state delivery".to_owned(),
230                ))
231            }
232            PlatformMakerEvent::MakerSnapshot {
233                schema_version,
234                contract_version,
235                market_id,
236                wallet_address,
237                stream_id,
238                sequence,
239                status,
240                fills,
241                ..
242            } => {
243                self.validate_identity(
244                    *schema_version,
245                    contract_version,
246                    market_id,
247                    wallet_address,
248                )?;
249                let next = validate_response_atoms(sequence, "sequence", false)?;
250                if stream_id != &self.stream_id || next <= self.sequence {
251                    return Err(SdkError::InvalidResponse(
252                        "maker recovery snapshot did not advance its sequence".to_owned(),
253                    ));
254                }
255                validate_maker_stream_state(status, fills, &self.market_id, &self.wallet_address)?;
256                self.sequence = next;
257            }
258            PlatformMakerEvent::MakerFill {
259                schema_version,
260                contract_version,
261                market_id,
262                wallet_address,
263                stream_id,
264                sequence,
265                previous_sequence,
266                fill,
267                ..
268            } => {
269                self.validate_identity(
270                    *schema_version,
271                    contract_version,
272                    market_id,
273                    wallet_address,
274                )?;
275                self.validate_sequence(stream_id, sequence, previous_sequence)?;
276                validate_maker_fills(std::slice::from_ref(fill))?;
277            }
278            PlatformMakerEvent::MakerStatus {
279                schema_version,
280                contract_version,
281                market_id,
282                wallet_address,
283                stream_id,
284                sequence,
285                previous_sequence,
286                status,
287                ..
288            } => {
289                self.validate_identity(
290                    *schema_version,
291                    contract_version,
292                    market_id,
293                    wallet_address,
294                )?;
295                self.validate_sequence(stream_id, sequence, previous_sequence)?;
296                validate_maker_stream_state(status, &[], &self.market_id, &self.wallet_address)?;
297            }
298            PlatformMakerEvent::Heartbeat {
299                schema_version,
300                contract_version,
301                market_id,
302                wallet_address,
303                stream_id,
304                sequence,
305                previous_sequence,
306                ..
307            } => {
308                self.validate_identity(
309                    *schema_version,
310                    contract_version,
311                    market_id,
312                    wallet_address,
313                )?;
314                self.validate_sequence(stream_id, sequence, previous_sequence)?;
315            }
316        }
317        Ok(())
318    }
319
320    fn validate_identity(
321        &self,
322        schema_version: u16,
323        contract_version: &str,
324        market_id: &str,
325        wallet_address: &str,
326    ) -> Result<(), SdkError> {
327        validate_maker_identity(
328            schema_version,
329            contract_version,
330            market_id,
331            wallet_address,
332            &self.market_id,
333            &self.wallet_address,
334        )
335    }
336
337    fn validate_sequence(
338        &mut self,
339        stream_id: &str,
340        sequence: &str,
341        previous_sequence: &str,
342    ) -> Result<(), SdkError> {
343        let next = validate_response_atoms(sequence, "sequence", false)?;
344        let previous = validate_response_atoms(previous_sequence, "previous_sequence", false)?;
345        if stream_id != self.stream_id
346            || previous != self.sequence
347            || next != previous.saturating_add(1)
348        {
349            return Err(SdkError::InvalidResponse(
350                "maker stream sequence gap detected".to_owned(),
351            ));
352        }
353        self.sequence = next;
354        Ok(())
355    }
356}
357
358async fn next_text(socket: &mut PlatformSocket, expected: &str) -> Result<String, SdkError> {
359    let frame = tokio::time::timeout(AUTH_TIMEOUT, socket.next())
360        .await
361        .map_err(|_| SdkError::Stream(format!("{expected} timed out")))?
362        .ok_or_else(|| SdkError::Stream(format!("socket closed before {expected}")))?
363        .map_err(|error| SdkError::Stream(error.to_string()))?;
364    let Message::Text(text) = frame else {
365        return Err(SdkError::InvalidResponse(format!(
366            "expected a text {expected}"
367        )));
368    };
369    Ok(text.to_string())
370}
371
372fn validate_maker_identity(
373    schema_version: u16,
374    contract_version: &str,
375    actual_market: &str,
376    actual_wallet: &str,
377    expected_market: &str,
378    expected_wallet: &str,
379) -> Result<(), SdkError> {
380    validate_platform_market_response(
381        schema_version,
382        contract_version,
383        actual_market,
384        expected_market,
385    )?;
386    if actual_wallet != expected_wallet {
387        return Err(SdkError::InvalidResponse(
388            "maker stream wallet does not match signed request".to_owned(),
389        ));
390    }
391    Ok(())
392}
393
394/// The status inside a stream event must be the same owner-scoped projection
395/// the HTTP read returns, bound to the same market and wallet.
396fn validate_maker_stream_state(
397    status: &PlatformMakerStatusResponse,
398    fills: &[PlatformMakerFill],
399    market_id: &str,
400    wallet_address: &str,
401) -> Result<(), SdkError> {
402    validate_platform_market_response(
403        status.schema_version,
404        &status.contract_version,
405        &status.market_id,
406        market_id,
407    )?;
408    if status.wallet_address != wallet_address {
409        return Err(SdkError::InvalidResponse(
410            "maker stream status wallet does not match the stream".to_owned(),
411        ));
412    }
413    validate_maker_status(status)?;
414    validate_maker_fills(fills)
415}
416
417pub(crate) fn validate_maker_fills(fills: &[PlatformMakerFill]) -> Result<(), SdkError> {
418    if fills.len() > 2_000 {
419        return Err(SdkError::InvalidResponse(
420            "maker stream fills exceed the bounded size".to_owned(),
421        ));
422    }
423    let projected: Vec<PlatformAccountFill> = fills
424        .iter()
425        .map(|fill| PlatformAccountFill {
426            fill_id: fill.fill_id.clone(),
427            side: fill.side,
428            price_atoms: fill.price_atoms.clone(),
429            size_atoms: fill.size_atoms.clone(),
430            fee_quote_atoms: fill.fee_quote_atoms.clone(),
431            fee_is_final: fill.fee_is_final,
432            settlement: fill.settlement,
433            executed_at_ms: fill.executed_at_ms,
434            confirmed_at_ms: fill.confirmed_at_ms,
435            transaction_id: fill.transaction_id.clone(),
436            realized_pnl_quote_atoms: fill.realized_pnl_quote_atoms.clone(),
437        })
438        .collect();
439    validate_account_state(&[], &projected)
440        .map_err(|_| SdkError::InvalidResponse("maker stream contains an invalid fill".to_owned()))
441}
442
443pub fn maker_stream_auth_message(
444    market_id: &str,
445    wallet_address: &str,
446    challenge: &str,
447) -> Result<Vec<u8>, SdkError> {
448    let market_id = validate_platform_market_id(market_id)?;
449    let wallet_address = canonical_public_key(wallet_address, "wallet_address")?;
450    if challenge.len() != 64
451        || !challenge
452            .bytes()
453            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
454    {
455        return Err(SdkError::InvalidRequest(
456            "maker stream challenge must be 32-byte lowercase hexadecimal".to_owned(),
457        ));
458    }
459    Ok(
460        format!("{MAKER_STREAM_AUTH_DOMAIN}\n{market_id}\n{wallet_address}\n{challenge}")
461            .into_bytes(),
462    )
463}
strata_sdk/maker_stream.rs

strata_sdk/
maker_stream.rs
