feat: Streaming cross-chain quotes via SSE (--stream)

README.md

@@ -258,8 +258,18 @@ No gas fees required. The 0x protocol handles gas on your behalf.
 # Sort by fastest bridge
 0x cross-chain --from base --to arbitrum \
   --sell 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913 --buy 0xaf88d065e77c8cC2239327C5EDb3A432268e5831 --amount 1000000 --sort speed
+
+# Stream quotes as they're discovered (SSE)
+0x cross-chain --from base --to arbitrum \
+  --sell 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913 --buy 0xaf88d065e77c8cC2239327C5EDb3A432268e5831 --amount 1000000 --stream
 ```
 
+`--stream` fetches quotes progressively via Server-Sent Events instead of
+waiting for all bridges to respond. Progress appears on stderr; the final
+output is identical to the non-streaming mode, with quotes re-sorted best
+price first before selection. Cannot be combined with `--sort`; `--max-quotes`
+defaults to 5 (1-10) instead of 3.
+
 Solana-origin swaps automatically include routes that need an extra one-shot
 transaction signer (e.g. Circle CCTP): the CLI generates a fresh keypair per
 quote request, sends its pubkey with the quote, and co-signs at submission.

src/api/cross_chain.rs

@@ -90,6 +90,134 @@ pub struct CrossChainAllowance {
     pub spender: String,
 }
 
+/// One decoded SSE message from `GET /cross-chain/quotes/stream`. Each `data:`
+/// line is a JSON object of this shape: a `zid` correlation id plus a tagged
+/// `event`.
+#[derive(Debug, Clone, Deserialize)]
+pub struct CrossChainStreamMessage {
+    #[serde(default)]
+    pub zid: Option<String>,
+    pub event: CrossChainStreamEvent,
+}
+
+/// The `event` payload, discriminated by its inner `type` field.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum CrossChainStreamEvent {
+    /// A discovered quote. Arrives zero or more times before the terminal
+    /// `result`. Boxed to keep the enum small — the quote payload dwarfs the
+    /// other variants.
+    Quote {
+        data: Box<CrossChainStreamQuoteData>,
+    },
+    /// Terminal event: the stream is done. `liquidityAvailable` mirrors the
+    /// buffered response's field.
+    Result { data: CrossChainStreamResultData },
+    /// Terminal error event.
+    Error { data: CrossChainStreamErrorData },
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CrossChainStreamQuoteData {
+    pub quote: CrossChainQuote,
+    #[serde(default)]
+    pub allowance_target: Option<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CrossChainStreamResultData {
+    #[serde(default)]
+    pub liquidity_available: bool,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct CrossChainStreamErrorData {
+    #[serde(default)]
+    pub message: Option<String>,
+    #[serde(default)]
+    pub code: Option<String>,
+}
+
+/// Incremental Server-Sent Events parser (WHATWG subset). Feed raw byte
+/// chunks as they arrive off the wire; complete events are dispatched at the
+/// blank-line boundary, with consecutive `data:` lines joined by `\n` per the
+/// spec. `\r\n`, `\n`, and bare `\r` all terminate lines — including a `\r\n`
+/// pair split across two chunks. Comments (`:`-prefixed) and non-`data`
+/// fields (`event:`, `id:`, `retry:`) are ignored — the streaming quotes
+/// endpoint carries everything in `data:`.
+///
+/// Bytes are buffered raw and only decoded once a line is complete, so a
+/// multi-byte UTF-8 sequence split across two chunks is never mangled.
+#[derive(Default)]
+pub struct SseParser {
+    line: Vec<u8>,
+    data_lines: Vec<String>,
+    /// The previous byte was a bare `\r` line terminator; a `\n` immediately
+    /// after belongs to the same `\r\n` pair and must be swallowed.
+    skip_next_lf: bool,
+}
+
+impl SseParser {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Feed a chunk of bytes; returns the joined `data:` payload of every
+    /// event completed (blank-line-terminated) by this chunk.
+    pub fn feed(&mut self, chunk: &[u8]) -> Vec<String> {
+        let mut out = Vec::new();
+        for &b in chunk {
+            if std::mem::take(&mut self.skip_next_lf) && b == b'\n' {
+                continue;
+            }
+            match b {
+                b'\n' => self.end_line(&mut out),
+                b'\r' => {
+                    self.end_line(&mut out);
+                    self.skip_next_lf = true;
+                }
+                _ => self.line.push(b),
+            }
+        }
+        out
+    }
+
+    /// Flush at end-of-stream: a final event whose terminating blank line
+    /// never arrived is returned rather than silently dropped.
+    pub fn finish(&mut self) -> Option<String> {
+        if !self.line.is_empty() {
+            // Complete the dangling line. Only a blank line dispatches an
+            // event, so `out` stays empty here; any `data:` content lands in
+            // `data_lines`.
+            let mut out = Vec::new();
+            self.end_line(&mut out);
+        }
+        if self.data_lines.is_empty() {
+            None
+        } else {
+            Some(std::mem::take(&mut self.data_lines).join("\n"))
+        }
+    }
+
+    fn end_line(&mut self, out: &mut Vec<String>) {
+        let line = String::from_utf8_lossy(&self.line).into_owned();
+        self.line.clear();
+        if line.is_empty() {
+            // Event boundary: dispatch accumulated data, if any.
+            if !self.data_lines.is_empty() {
+                out.push(std::mem::take(&mut self.data_lines).join("\n"));
+            }
+        } else if let Some(rest) = line.strip_prefix("data:") {
+            // Honor the optional single space after the colon per the spec.
+            self.data_lines
+                .push(rest.strip_prefix(' ').unwrap_or(rest).to_string());
+        }
+        // Comments (`:`-prefixed) and other fields are ignored.
+    }
+}
+
 /// Response from GET /cross-chain/status
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -178,10 +306,25 @@ pub struct CrossChainQuoteParams<'a> {
 }
 
 impl CrossChainQuoteParams<'_> {
-    /// Build the query-string pairs for the request. Returns owned values so
-    /// the caller can borrow them into `&[(&str, &str)]`. Optional filters are
-    /// only included when non-empty; comma-separated lists are joined here.
+    /// Build the query-string pairs for the buffered `/cross-chain/quotes`
+    /// request. Returns owned values so the caller can borrow them into
+    /// `&[(&str, &str)]`.
     fn query_pairs(&self) -> Vec<(&'static str, String)> {
+        self.build_query_pairs(true)
+    }
+
+    /// Build the query-string pairs for the `/cross-chain/quotes/stream`
+    /// request. Identical to [`Self::query_pairs`] except `sortQuotesBy` is
+    /// omitted — the streaming endpoint doesn't support it (quotes arrive in
+    /// discovery order).
+    fn stream_query_pairs(&self) -> Vec<(&'static str, String)> {
+        self.build_query_pairs(false)
+    }
+
+    /// Shared query-pair builder. `include_sort` gates `sortQuotesBy`, the one
+    /// parameter the streaming endpoint rejects. Optional filters are only
+    /// included when non-empty; comma-separated lists are joined here.
+    fn build_query_pairs(&self, include_sort: bool) -> Vec<(&'static str, String)> {
         let mut params: Vec<(&'static str, String)> = vec![
             ("originChain", self.origin_chain.to_string()),
             ("destinationChain", self.destination_chain.to_string()),
@@ -191,9 +334,11 @@ impl CrossChainQuoteParams<'_> {
             ("originAddress", self.origin_address.to_string()),
             ("destinationAddress", self.destination_address.to_string()),
             ("slippageBps", self.slippage_bps.unwrap_or(100).to_string()),
-            ("sortQuotesBy", self.sort_by.unwrap_or("price").to_string()),
             ("maxNumQuotes", self.max_quotes.unwrap_or(3).to_string()),
         ];
+        if include_sort {
+            params.push(("sortQuotesBy", self.sort_by.unwrap_or("price").to_string()));
+        }
         // Unlocks Solana-origin routes that need a one-shot extra signer
         // (e.g. CCTP's event account); the holder of the keypair must
         // co-sign the returned transaction.
@@ -226,6 +371,19 @@ impl ApiClient {
         self.get("/cross-chain/quotes", &borrowed).await
     }
 
+    /// Open the cross-chain quotes SSE stream. Returns the raw response for
+    /// chunked reading; the caller drives [`SseParser`] and decodes each
+    /// `data:` payload into a [`CrossChainStreamMessage`].
+    pub async fn open_cross_chain_quotes_stream(
+        &self,
+        params: &CrossChainQuoteParams<'_>,
+    ) -> Result<reqwest::Response, CliError> {
+        let owned = params.stream_query_pairs();
+        let borrowed: Vec<(&str, &str)> = owned.iter().map(|(k, v)| (*k, v.as_str())).collect();
+        self.get_stream("/cross-chain/quotes/stream", &borrowed)
+            .await
+    }
+
     /// Get cross-chain status
     pub async fn get_cross_chain_status(
         &self,
@@ -318,4 +476,181 @@ mod query_pairs_tests {
             Some("Uniswap_V3,Curve")
         );
     }
+
+    #[test]
+    fn stream_query_pairs_omits_sort() {
+        let empty: Vec<String> = Vec::new();
+        let params = base(&empty, &empty, &empty);
+        let pairs = params.stream_query_pairs();
+        assert!(value_for(&pairs, "sortQuotesBy").is_none());
+        // Everything else the buffered request sends is still present.
+        assert_eq!(value_for(&pairs, "maxNumQuotes"), Some("3"));
+        assert_eq!(value_for(&pairs, "originChain"), Some("8453"));
+    }
+}
+
+#[cfg(test)]
+mod sse_tests {
+    use super::*;
+
+    /// A minimal but structurally complete quote `data:` payload.
+    fn quote_json(buy: &str) -> String {
+        format!(
+            r#"{{"zid":"z1","event":{{"type":"quote","data":{{"quote":{{"sellAmount":"1000","buyAmount":"{buy}","minBuyAmount":"900","steps":[{{"type":"bridge","provider":"across"}}],"transaction":{{"chainType":"evm","details":{{"to":"0xabc","data":"0x"}}}},"gasCosts":null,"issues":null,"estimatedTimeSeconds":42,"quoteId":"q1"}},"allowanceTarget":"0xspender"}}}}}}"#
+        )
+    }
+
+    #[test]
+    fn parses_multiple_events_in_one_chunk() {
+        let mut parser = SseParser::new();
+        let chunk = format!(
+            "data: {}\n\ndata: {}\n\n",
+            quote_json("100"),
+            quote_json("200")
+        );
+        let payloads = parser.feed(chunk.as_bytes());
+        assert_eq!(payloads.len(), 2);
+        // The leading space after `data:` is stripped, leaving raw JSON.
+        assert!(payloads[0].starts_with('{'));
+    }
+
+    #[test]
+    fn reassembles_event_split_across_chunks() {
+        let mut parser = SseParser::new();
+        let full = format!("data: {}\n\n", quote_json("100"));
+        let split = full.len() / 2;
+        // No complete event yet — nothing emitted.
+        assert!(parser.feed(&full.as_bytes()[..split]).is_empty());
+        let payloads = parser.feed(&full.as_bytes()[split..]);
+        assert_eq!(payloads.len(), 1);
+        let msg: CrossChainStreamMessage = serde_json::from_str(&payloads[0]).unwrap();
+        assert!(matches!(msg.event, CrossChainStreamEvent::Quote { .. }));
+    }
+
+    #[test]
+    fn joins_consecutive_data_lines_with_newline() {
+        // Per the WHATWG spec, one event may span several `data:` lines; the
+        // dispatched payload joins them with `\n`.
+        let mut parser = SseParser::new();
+        let payloads = parser.feed(b"data: part1\ndata: part2\n\n");
+        assert_eq!(payloads, vec!["part1\npart2".to_string()]);
+    }
+
+    #[test]
+    fn handles_crlf_line_terminators() {
+        let mut parser = SseParser::new();
+        let payloads = parser.feed(b"data: X\r\n\r\ndata: Y\r\n\r\n");
+        assert_eq!(payloads, vec!["X".to_string(), "Y".to_string()]);
+    }
+
+    #[test]
+    fn handles_bare_cr_line_terminators() {
+        let mut parser = SseParser::new();
+        let payloads = parser.feed(b"data: X\r\rdata: Y\r\r");
+        assert_eq!(payloads, vec!["X".to_string(), "Y".to_string()]);
+    }
+
+    #[test]
+    fn handles_crlf_split_across_chunks() {
+        // A `\r` at the end of one chunk followed by `\n` at the start of the
+        // next is one terminator, not two.
+        let mut parser = SseParser::new();
+        assert!(parser.feed(b"data: X\r").is_empty());
+        let payloads = parser.feed(b"\n\r\n");
+        assert_eq!(payloads, vec!["X".to_string()]);
+    }
+
+    #[test]
+    fn finish_flushes_pending_event_at_end_of_stream() {
+        // The terminating blank line never arrived; end-of-stream still
+        // yields the accumulated data instead of dropping it.
+        let mut parser = SseParser::new();
+        assert!(parser.feed(b"data: X\n").is_empty());
+        assert_eq!(parser.finish().as_deref(), Some("X"));
+        // A second finish is a no-op.
+        assert!(parser.finish().is_none());
+    }
+
+    #[test]
+    fn finish_flushes_dangling_unterminated_line() {
+        let mut parser = SseParser::new();
+        assert!(parser.feed(b"data: X").is_empty());
+        assert_eq!(parser.finish().as_deref(), Some("X"));
+    }
+
+    #[test]
+    fn ignores_comments_blank_lines_and_other_fields() {
+        let mut parser = SseParser::new();
+        let chunk = format!(
+            ": keep-alive comment\n\nevent: quote\nid: 7\ndata: {}\n\n",
+            quote_json("100")
+        );
+        let payloads = parser.feed(chunk.as_bytes());
+        // Only the `data:` line contributes to the dispatched payload.
+        assert_eq!(payloads.len(), 1);
+        assert!(payloads[0].starts_with('{'));
+    }
+
+    // A `data:` payload whose JSON is malformed is deliberately *skipped* by
+    // the consuming loop rather than aborting the stream: the parser still
+    // emits the raw payload string, but decoding it fails and the caller
+    // drops it. This test pins that the parser itself is lossless (emits the
+    // payload) and that decode surfaces an error the caller can choose to
+    // ignore.
+    #[test]
+    fn malformed_json_payload_is_emitted_but_fails_to_decode() {
+        let mut parser = SseParser::new();
+        let payloads = parser.feed(b"data: {not valid json}\n\n");
+        assert_eq!(payloads.len(), 1);
+        assert!(serde_json::from_str::<CrossChainStreamMessage>(&payloads[0]).is_err());
+    }
+
+    #[test]
+    fn deserializes_quote_event() {
+        let msg: CrossChainStreamMessage = serde_json::from_str(&quote_json("12345")).unwrap();
+        assert_eq!(msg.zid.as_deref(), Some("z1"));
+        match msg.event {
+            CrossChainStreamEvent::Quote { data } => {
+                assert_eq!(data.quote.buy_amount, "12345");
+                assert_eq!(data.allowance_target.as_deref(), Some("0xspender"));
+                assert_eq!(data.quote.bridge_provider(), "across");
+            }
+            _ => panic!("expected quote event"),
+        }
+    }
+
+    #[test]
+    fn deserializes_result_event() {
+        let json = r#"{"zid":"z2","event":{"type":"result","data":{"liquidityAvailable":true}}}"#;
+        let msg: CrossChainStreamMessage = serde_json::from_str(json).unwrap();
+        match msg.event {
+            CrossChainStreamEvent::Result { data } => assert!(data.liquidity_available),
+            _ => panic!("expected result event"),
+        }
+    }
+
+    #[test]
+    fn deserializes_result_event_with_empty_data() {
+        // A terminal `result` missing `liquidityAvailable` still parses;
+        // assemble() then infers availability from the collected quotes.
+        let json = r#"{"zid":"z2","event":{"type":"result","data":{}}}"#;
+        let msg: CrossChainStreamMessage = serde_json::from_str(json).unwrap();
+        match msg.event {
+            CrossChainStreamEvent::Result { data } => assert!(!data.liquidity_available),
+            _ => panic!("expected result event"),
+        }
+    }
+
+    #[test]
+    fn deserializes_error_event() {
+        let json = r#"{"zid":"z3","event":{"type":"error","data":{"message":"boom","code":"NO_LIQUIDITY"}}}"#;
+        let msg: CrossChainStreamMessage = serde_json::from_str(json).unwrap();
+        match msg.event {
+            CrossChainStreamEvent::Error { data } => {
+                assert_eq!(data.message.as_deref(), Some("boom"));
+                assert_eq!(data.code.as_deref(), Some("NO_LIQUIDITY"));
+            }
+            _ => panic!("expected error event"),
+        }
+    }
 }