feat(ws): align auth and event_type parsing

src/decode.rs

@@ -432,44 +432,61 @@ impl Decoder<Market> for RawMarketResponse {
     }
 }
 
-/// WebSocket message parsing
-pub fn parse_stream_message(raw: &str) -> Result<StreamMessage> {
-    let value: Value = serde_json::from_str(raw)?;
+/// WebSocket message parsing (official `event_type` shape).
+///
+/// Polymarket WebSocket servers may send either a single JSON object or a batch array.
+/// This parser is tolerant:
+/// - Unknown/unsupported `event_type`s are ignored.
+/// - Invalid entries inside a batch are skipped (do not fail the whole batch).
+pub fn parse_stream_messages(raw: &str) -> Result<Vec<StreamMessage>> {
+    parse_stream_messages_bytes(raw.as_bytes())
+}
 
-    let msg_type = value["type"]
-        .as_str()
-        .ok_or_else(|| PolyfillError::parse("Missing message type".to_string(), None))?;
+/// See `parse_stream_messages`.
+pub fn parse_stream_messages_bytes(bytes: &[u8]) -> Result<Vec<StreamMessage>> {
+    let value: Value = serde_json::from_slice(bytes)?;
 
-    match msg_type {
-        "book_update" => {
-            let data = value["data"].clone();
-            let delta: OrderDelta = serde_json::from_value(data)?;
-            Ok(StreamMessage::BookUpdate { data: delta })
-        },
-        "trade" => {
-            let data = value["data"].clone();
-            let raw_trade: RawTradeResponse = serde_json::from_value(data)?;
-            let fill = raw_trade.decode()?;
-            Ok(StreamMessage::Trade { data: fill })
-        },
-        "order_update" => {
-            let data = value["data"].clone();
-            let raw_order: RawOrderResponse = serde_json::from_value(data)?;
-            let order = raw_order.decode()?;
-            Ok(StreamMessage::OrderUpdate { data: order })
-        },
-        "heartbeat" => {
-            let timestamp = value["timestamp"]
-                .as_str()
-                .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
-                .map(|dt| dt.with_timezone(&Utc))
-                .unwrap_or_else(Utc::now);
-            Ok(StreamMessage::Heartbeat { timestamp })
-        },
-        _ => Err(PolyfillError::parse(
-            format!("Unknown message type: {}", msg_type),
-            None,
-        )),
+    match value {
+        Value::Object(map) => {
+            let event_type = map.get("event_type").and_then(Value::as_str);
+            match event_type {
+                None => Ok(vec![]),
+                Some(_) => {
+                    let msg: StreamMessage = serde_json::from_value(Value::Object(map))?;
+                    match msg {
+                        StreamMessage::Unknown => Ok(vec![]),
+                        other => Ok(vec![other]),
+                    }
+                }
+            }
+        }
+        Value::Array(arr) => Ok(arr
+            .into_iter()
+            .filter_map(|elem| {
+                let obj = elem.as_object()?;
+                let event_type = obj.get("event_type").and_then(Value::as_str)?;
+                // Skip unknown event types early (forward compatibility).
+                match event_type {
+                    "book"
+                    | "price_change"
+                    | "tick_size_change"
+                    | "last_trade_price"
+                    | "best_bid_ask"
+                    | "new_market"
+                    | "market_resolved"
+                    | "trade"
+                    | "order" => {}
+                    _ => return None,
+                }
+
+                match serde_json::from_value::<StreamMessage>(Value::Object(obj.clone())) {
+                    Ok(StreamMessage::Unknown) => None,
+                    Ok(msg) => Some(msg),
+                    Err(_) => None,
+                }
+            })
+            .collect()),
+        _ => Ok(vec![]),
     }
 }
 

src/stream.rs

@@ -274,12 +274,12 @@ impl WebSocketStream {
             tokio_tungstenite::tungstenite::Message::Text(text) => {
                 debug!("Received WebSocket message: {}", text);
 
-                // Parse the message according to Polymarket's format
-                let stream_message = self.parse_polymarket_message(&text)?;
-
-                // Send to internal channel
-                if let Err(e) = self.tx.send(stream_message) {
-                    error!("Failed to send message to internal channel: {}", e);
+                // Parse the message according to Polymarket's `event_type` format
+                let stream_messages = crate::decode::parse_stream_messages(&text)?;
+                for stream_message in stream_messages {
+                    if let Err(e) = self.tx.send(stream_message) {
+                        error!("Failed to send message to internal channel: {}", e);
+                    }
                 }
 
                 self.stats.messages_received += 1;
@@ -313,115 +313,10 @@ impl WebSocketStream {
         Ok(())
     }
 
-    /// Parse Polymarket WebSocket message format
+    /// Parse Polymarket WebSocket message(s) in `event_type` format.
     #[allow(dead_code)]
-    fn parse_polymarket_message(&self, text: &str) -> Result<StreamMessage> {
-        let value: Value = serde_json::from_str(text).map_err(|e| {
-            PolyfillError::parse(
-                format!("Failed to parse WebSocket message: {}", e),
-                Some(Box::new(e)),
-            )
-        })?;
-
-        // Extract message type
-        let message_type = value.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
-            PolyfillError::parse("Missing 'type' field in WebSocket message", None)
-        })?;
-
-        match message_type {
-            "book_update" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse book update: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::BookUpdate { data })
-            },
-            "trade" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse trade: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::Trade { data })
-            },
-            "order_update" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse order update: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::OrderUpdate { data })
-            },
-            "user_order_update" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse user order update: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::UserOrderUpdate { data })
-            },
-            "user_trade" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse user trade: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::UserTrade { data })
-            },
-            "market_book_update" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse market book update: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::MarketBookUpdate { data })
-            },
-            "market_trade" => {
-                let data =
-                    serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
-                        .map_err(|e| {
-                            PolyfillError::parse(
-                                format!("Failed to parse market trade: {}", e),
-                                Some(Box::new(e)),
-                            )
-                        })?;
-                Ok(StreamMessage::MarketTrade { data })
-            },
-            "heartbeat" => {
-                let timestamp = value
-                    .get("timestamp")
-                    .and_then(|v| v.as_u64())
-                    .map(|ts| chrono::DateTime::from_timestamp(ts as i64, 0).unwrap_or_default())
-                    .unwrap_or_else(Utc::now);
-                Ok(StreamMessage::Heartbeat { timestamp })
-            },
-            _ => {
-                warn!("Unknown message type: {}", message_type);
-                // Return heartbeat as fallback
-                Ok(StreamMessage::Heartbeat {
-                    timestamp: Utc::now(),
-                })
-            },
-        }
+    fn parse_polymarket_messages(&self, text: &str) -> Result<Vec<StreamMessage>> {
+        crate::decode::parse_stream_messages(text)
     }
 
     /// Reconnect with exponential backoff
@@ -476,33 +371,58 @@ impl Stream for WebSocketStream {
     type Item = Result<StreamMessage>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        // First check internal channel
-        if let Poll::Ready(Some(message)) = self.rx.poll_recv(cx) {
-            return Poll::Ready(Some(Ok(message)));
-        }
+        loop {
+            // First drain any parsed messages
+            if let Poll::Ready(Some(message)) = self.rx.poll_recv(cx) {
+                return Poll::Ready(Some(Ok(message)));
+            }
+
+            let Some(connection) = &mut self.connection else {
+                return Poll::Ready(None);
+            };
 
-        // Then check WebSocket connection
-        if let Some(connection) = &mut self.connection {
             match connection.poll_next_unpin(cx) {
-                Poll::Ready(Some(Ok(_message))) => {
-                    // Simplified message handling
-                    Poll::Ready(Some(Ok(StreamMessage::Heartbeat {
-                        timestamp: Utc::now(),
-                    })))
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Some(Ok(ws_message))) => match ws_message {
+                    tokio_tungstenite::tungstenite::Message::Text(text) => {
+                        match crate::decode::parse_stream_messages(&text) {
+                            Ok(messages) => {
+                                for msg in messages {
+                                    let _ = self.tx.send(msg);
+                                }
+                                self.stats.messages_received += 1;
+                                self.stats.last_message_time = Some(Utc::now());
+                                continue;
+                            }
+                            Err(e) => {
+                                self.stats.errors += 1;
+                                return Poll::Ready(Some(Err(e)));
+                            }
+                        }
+                    }
+                    tokio_tungstenite::tungstenite::Message::Close(_) => {
+                        info!("WebSocket connection closed by server");
+                        self.connection = None;
+                        return Poll::Ready(None);
+                    }
+                    tokio_tungstenite::tungstenite::Message::Ping(_) => {
+                        // Best-effort: tokio-tungstenite/tungstenite may handle pings internally.
+                        continue;
+                    }
+                    tokio_tungstenite::tungstenite::Message::Pong(_) => continue,
+                    tokio_tungstenite::tungstenite::Message::Binary(_) => continue,
+                    tokio_tungstenite::tungstenite::Message::Frame(_) => continue,
                 },
                 Poll::Ready(Some(Err(e))) => {
                     error!("WebSocket error: {}", e);
                     self.stats.errors += 1;
-                    Poll::Ready(Some(Err(e.into())))
-                },
+                    return Poll::Ready(Some(Err(e.into())));
+                }
                 Poll::Ready(None) => {
                     info!("WebSocket stream ended");
-                    Poll::Ready(None)
-                },
-                Poll::Pending => Poll::Pending,
+                    return Poll::Ready(None);
+                }
             }
-        } else {
-            Poll::Ready(None)
         }
     }
 }
@@ -655,19 +575,19 @@ mod tests {
         let mut stream = MockStream::new();
 
         // Add some test messages
-        stream.add_message(StreamMessage::Heartbeat {
-            timestamp: Utc::now(),
-        });
-        stream.add_message(StreamMessage::BookUpdate {
-            data: OrderDelta {
-                token_id: "test".to_string(),
-                timestamp: Utc::now(),
-                side: Side::BUY,
-                price: rust_decimal_macros::dec!(0.5),
-                size: rust_decimal_macros::dec!(100),
-                sequence: 1,
-            },
-        });
+        stream.add_message(StreamMessage::Book(BookUpdate {
+            asset_id: "1".to_string(),
+            market: "0xabc".to_string(),
+            timestamp: 1_234_567_890,
+            bids: vec![],
+            asks: vec![],
+            hash: None,
+        }));
+        stream.add_message(StreamMessage::PriceChange(PriceChange {
+            market: "0xabc".to_string(),
+            timestamp: 1_234_567_891,
+            price_changes: vec![],
+        }));
 
         assert!(stream.is_connected());
         assert_eq!(stream.get_stats().messages_received, 2);
@@ -680,9 +600,14 @@ mod tests {
         manager.add_stream(mock_stream);
 
         // Test message broadcasting
-        let message = StreamMessage::Heartbeat {
-            timestamp: Utc::now(),
-        };
+        let message = StreamMessage::Book(BookUpdate {
+            asset_id: "1".to_string(),
+            market: "0xabc".to_string(),
+            timestamp: 1_234_567_890,
+            bids: vec![],
+            asks: vec![],
+            hash: None,
+        });
         assert!(manager.broadcast_message(message).is_ok());
     }
 }

src/types.rs

@@ -659,18 +659,11 @@ impl Default for ClientConfig {
     }
 }
 
-/// WebSocket authentication for Polymarket API
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct WssAuth {
-    /// User's Ethereum address
-    pub address: String,
-    /// EIP-712 signature
-    pub signature: String,
-    /// Unix timestamp
-    pub timestamp: u64,
-    /// Nonce for replay protection
-    pub nonce: String,
-}
+/// WebSocket authentication for Polymarket API user channel.
+///
+/// Polymarket's CLOB WebSocket expects the same L2 API credentials used for HTTP calls:
+/// `{ apiKey, secret, passphrase }`.
+pub type WssAuth = ApiCredentials;
 
 /// WebSocket subscription request
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -699,28 +692,247 @@ pub struct WssSubscription {
     pub auth: Option<WssAuth>,
 }
 
-/// WebSocket message types for streaming
+/// WebSocket message types for streaming (official Polymarket `event_type` format).
 #[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(tag = "type")]
+#[serde(tag = "event_type")]
 pub enum StreamMessage {
-    #[serde(rename = "book_update")]
-    BookUpdate { data: OrderDelta },
+    /// Full or incremental orderbook update
+    #[serde(rename = "book")]
+    Book(BookUpdate),
+    /// Price change notification (single or batched)
+    #[serde(rename = "price_change")]
+    PriceChange(PriceChange),
+    /// Tick size change notification
+    #[serde(rename = "tick_size_change")]
+    TickSizeChange(TickSizeChange),
+    /// Last trade price update
+    #[serde(rename = "last_trade_price")]
+    LastTradePrice(LastTradePrice),
+    /// Best bid/ask update (requires `custom_feature_enabled`)
+    #[serde(rename = "best_bid_ask")]
+    BestBidAsk(BestBidAsk),
+    /// New market created (requires `custom_feature_enabled`)
+    #[serde(rename = "new_market")]
+    NewMarket(NewMarket),
+    /// Market resolved (requires `custom_feature_enabled`)
+    #[serde(rename = "market_resolved")]
+    MarketResolved(MarketResolved),
+    /// User trade execution (authenticated channel)
     #[serde(rename = "trade")]
-    Trade { data: FillEvent },
-    #[serde(rename = "order_update")]
-    OrderUpdate { data: Order },
-    #[serde(rename = "heartbeat")]
-    Heartbeat { timestamp: DateTime<Utc> },
-    /// User channel events
-    #[serde(rename = "user_order_update")]
-    UserOrderUpdate { data: Order },
-    #[serde(rename = "user_trade")]
-    UserTrade { data: FillEvent },
-    /// Market channel events
-    #[serde(rename = "market_book_update")]
-    MarketBookUpdate { data: OrderDelta },
-    #[serde(rename = "market_trade")]
-    MarketTrade { data: FillEvent },
+    Trade(TradeMessage),
+    /// User order update (authenticated channel)
+    #[serde(rename = "order")]
+    Order(OrderMessage),
+    /// Forward-compatible catch-all for new/unknown event types.
+    #[serde(other)]
+    Unknown,
+}
+
+/// Orderbook update message (full snapshot or delta).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BookUpdate {
+    pub asset_id: String,
+    pub market: String,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+    #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")]
+    pub bids: Vec<OrderSummary>,
+    #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")]
+    pub asks: Vec<OrderSummary>,
+    #[serde(default)]
+    pub hash: Option<String>,
+}
+
+/// Unified wire format for `price_change` events.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PriceChange {
+    pub market: String,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+    #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")]
+    pub price_changes: Vec<PriceChangeEntry>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PriceChangeEntry {
+    pub asset_id: String,
+    pub price: Decimal,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub size: Option<Decimal>,
+    pub side: Side,
+    #[serde(default)]
+    pub hash: Option<String>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub best_bid: Option<Decimal>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub best_ask: Option<Decimal>,
+}
+
+/// Tick size change event.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TickSizeChange {
+    pub asset_id: String,
+    pub market: String,
+    pub old_tick_size: Decimal,
+    pub new_tick_size: Decimal,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+}
+
+/// Last trade price update.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LastTradePrice {
+    pub asset_id: String,
+    pub market: String,
+    pub price: Decimal,
+    #[serde(default)]
+    pub side: Option<Side>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub size: Option<Decimal>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub fee_rate_bps: Option<Decimal>,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+}
+
+/// Best bid/ask update.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BestBidAsk {
+    pub market: String,
+    pub asset_id: String,
+    pub best_bid: Decimal,
+    pub best_ask: Decimal,
+    pub spread: Decimal,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+}
+
+/// New market created event.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct NewMarket {
+    pub id: String,
+    pub question: String,
+    pub market: String,
+    pub slug: String,
+    pub description: String,
+    #[serde(rename = "assets_ids", alias = "asset_ids")]
+    pub asset_ids: Vec<String>,
+    #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")]
+    pub outcomes: Vec<String>,
+    #[serde(default)]
+    pub event_message: Option<EventMessage>,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+}
+
+/// Market resolved event.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MarketResolved {
+    pub id: String,
+    #[serde(default)]
+    pub question: Option<String>,
+    pub market: String,
+    #[serde(default)]
+    pub slug: Option<String>,
+    #[serde(default)]
+    pub description: Option<String>,
+    #[serde(rename = "assets_ids", alias = "asset_ids")]
+    pub asset_ids: Vec<String>,
+    #[serde(default, deserialize_with = "crate::decode::deserializers::vec_from_null")]
+    pub outcomes: Vec<String>,
+    pub winning_asset_id: String,
+    pub winning_outcome: String,
+    #[serde(default)]
+    pub event_message: Option<EventMessage>,
+    #[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
+    pub timestamp: u64,
+}
+
+/// Event message object for market events.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct EventMessage {
+    pub id: String,
+    pub ticker: String,
+    pub slug: String,
+    pub title: String,
+    pub description: String,
+}
+
+/// User trade execution message.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TradeMessage {
+    pub id: String,
+    pub market: String,
+    pub asset_id: String,
+    pub side: Side,
+    pub size: Decimal,
+    pub price: Decimal,
+    #[serde(default)]
+    pub status: Option<String>,
+    #[serde(rename = "type", default)]
+    pub msg_type: Option<String>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_number_from_string"
+    )]
+    pub last_update: Option<u64>,
+    #[serde(
+        default,
+        alias = "match_time",
+        deserialize_with = "crate::decode::deserializers::optional_number_from_string"
+    )]
+    pub matchtime: Option<u64>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_number_from_string"
+    )]
+    pub timestamp: Option<u64>,
+}
+
+/// User order update message.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct OrderMessage {
+    pub id: String,
+    pub market: String,
+    pub asset_id: String,
+    pub side: Side,
+    pub price: Decimal,
+    #[serde(rename = "type", default)]
+    pub msg_type: Option<String>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub original_size: Option<Decimal>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_decimal_from_string"
+    )]
+    pub size_matched: Option<Decimal>,
+    #[serde(
+        default,
+        deserialize_with = "crate::decode::deserializers::optional_number_from_string"
+    )]
+    pub timestamp: Option<u64>,
+    #[serde(default)]
+    pub associate_trades: Option<Vec<String>>,
+    #[serde(default)]
+    pub status: Option<String>,
 }
 
 /// Subscription parameters for streaming