Fix market visibility subscription bugs (#365)

Author: crthpl

backend/.sqlx/query-04bf4b5be629f05d2b61c0bdb92b3b93fac1e86de0267d20a11483cbca7bfc3a.json

@@ -1196,6 +1196,44 @@ impl DB {
         Ok(is_visible)
     }
 
+    /// Check if a market is visible to any of the given accounts.
+    /// Returns true if the market has no visibility restrictions,
+    /// or if any of the owned accounts is in the `visible_to` list.
+    #[instrument(err, skip(self))]
+    pub async fn is_market_visible_to_any(
+        &self,
+        market_id: i64,
+        owned_accounts: &[i64],
+    ) -> SqlxResult<bool> {
+        // First check if the market has any restrictions at all
+        let has_restrictions = sqlx::query_scalar!(
+            r#"SELECT EXISTS (SELECT 1 FROM market_visible_to WHERE market_id = ?) as "exists!: bool""#,
+            market_id
+        )
+        .fetch_one(&self.pool)
+        .await?;
+
+        if !has_restrictions {
+            return Ok(true);
+        }
+
+        // Check if any owned account is in the visible_to list
+        for &account_id in owned_accounts {
+            let is_listed = sqlx::query_scalar!(
+                r#"SELECT EXISTS (SELECT 1 FROM market_visible_to WHERE market_id = ? AND account_id = ?) as "exists!: bool""#,
+                market_id,
+                account_id
+            )
+            .fetch_one(&self.pool)
+            .await?;
+            if is_listed {
+                return Ok(true);
+            }
+        }
+
+        Ok(false)
+    }
+
     #[must_use]
     pub fn get_all_live_orders(&self) -> BoxStream<'_, SqlxResult<Order>> {
         sqlx::query_as!(

backend/.sqlx/query-bbb71f664c418f08aa1b6253faeecc4cb5505e532603de6ee420b01487200372.json

@@ -121,6 +121,10 @@ async fn handle_socket_fallible(mut socket: WebSocket, app_state: AppState) -> a
                 match msg {
                     Ok(mut msg) => {
                         if !is_admin || !sudo_enabled {
+                            // Filter out messages about markets the user can't see
+                            if should_filter_for_visibility(db, &owned_accounts, &msg).await? {
+                                continue;
+                            }
                             conditionally_hide_user_ids(db, &owned_accounts, &mut msg).await?;
                         }
                         socket.send(msg.encode_to_vec().into()).await?;
@@ -502,6 +506,32 @@ async fn conditionally_hide_user_ids(
     Ok(())
 }
 
+/// Extract `market_id` from a broadcast message, if it's a market-related message.
+fn broadcast_market_id(msg: &ServerMessage) -> Option<i64> {
+    match &msg.message {
+        Some(SM::Market(m)) => Some(m.id),
+        Some(SM::MarketSettled(ms)) => Some(ms.id),
+        Some(SM::OrderCreated(oc)) => Some(oc.market_id),
+        Some(SM::OrdersCancelled(oc)) => Some(oc.market_id),
+        Some(SM::Redeemed(r)) => Some(r.fund_id),
+        _ => None,
+    }
+}
+
+/// Check if a broadcast message should be filtered out due to market visibility restrictions.
+/// Returns true if the message should be SKIPPED (not sent to this client).
+async fn should_filter_for_visibility(
+    db: &DB,
+    owned_accounts: &[i64],
+    msg: &ServerMessage,
+) -> anyhow::Result<bool> {
+    let Some(market_id) = broadcast_market_id(msg) else {
+        return Ok(false);
+    };
+    let is_visible = db.is_market_visible_to_any(market_id, owned_accounts).await?;
+    Ok(!is_visible)
+}
+
 struct ActAsInfo {
     request_id: String,
     account_id: i64,
@@ -624,15 +654,8 @@ async fn handle_client_message(
                 .await?
             {
                 Ok(market) => {
-                    let visible_to = market.visible_to.clone();
                     let msg = server_message(request_id, SM::Market(market.into()));
-                    if visible_to.is_empty() {
-                        subscriptions.send_public(msg);
-                    } else {
-                        for account_id in visible_to {
-                            subscriptions.send_private(account_id, msg.encode_to_vec().into());
-                        }
-                    }
+                    subscriptions.send_public(msg);
                 }
                 Err(failure) => {
                     fail!("CreateMarket", failure.message());
@@ -645,16 +668,10 @@ async fn handle_client_message(
                 Ok(db::MarketSettledWithAffectedAccounts {
                     market_settled,
                     affected_accounts,
-                    visible_to,
+                    ..
                 }) => {
                     let msg = server_message(request_id, SM::MarketSettled(market_settled.into()));
-                    if visible_to.is_empty() {
-                        subscriptions.send_public(msg);
-                    } else {
-                        for account_id in visible_to {
-                            subscriptions.send_private(account_id, msg.encode_to_vec().into());
-                        }
-                    }
+                    subscriptions.send_public(msg);
                     for account in affected_accounts {
                         subscriptions.notify_portfolio(account);
                     }
@@ -882,15 +899,8 @@ async fn handle_client_message(
 
             match db.edit_market(edit_market).await? {
                 Ok(market) => {
-                    let visible_to = market.visible_to.clone();
                     let msg = server_message(request_id, SM::Market(market.into()));
-                    if visible_to.is_empty() {
-                        subscriptions.send_public(msg);
-                    } else {
-                        for account_id in visible_to {
-                            subscriptions.send_private(account_id, msg.encode_to_vec().into());
-                        }
-                    }
+                    subscriptions.send_public(msg);
                 }
                 Err(err) => {
                     fail!("EditMarket", err.message());

backend/src/db.rs

@@ -20,8 +20,8 @@ use crate::{
     subscriptions::Subscriptions,
     websocket_api::{
         client_message::Message as CM, server_message::Message as SM, ActAs, Authenticate,
-        ClientMessage, CreateMarket, CreateOrder, GetFullTradeHistory,
-        RevokeOwnership, ServerMessage, SetSudo, Side,
+        ClientMessage, CreateMarket, CreateOrder, EditMarket, GetFullTradeHistory,
+        RevokeOwnership, ServerMessage, SettleMarket, SetSudo, Side,
     },
     AppState,
 };
@@ -259,6 +259,21 @@ impl TestClient {
         min_settlement: f64,
         max_settlement: f64,
         hide_account_ids: bool,
+    ) -> anyhow::Result<ServerMessage> {
+        self.create_market_with_visible_to(name, min_settlement, max_settlement, hide_account_ids, vec![]).await
+    }
+
+    /// Send a `CreateMarket` message with visibility restrictions.
+    ///
+    /// # Errors
+    /// Returns an error if sending fails.
+    pub async fn create_market_with_visible_to(
+        &mut self,
+        name: &str,
+        min_settlement: f64,
+        max_settlement: f64,
+        hide_account_ids: bool,
+        visible_to: Vec<i64>,
     ) -> anyhow::Result<ServerMessage> {
         let request_id = self.next_request_id();
         let msg = ClientMessage {
@@ -271,7 +286,7 @@ impl TestClient {
                 redeemable_for: vec![],
                 redeem_fee: 0.0,
                 hide_account_ids,
-                visible_to: vec![],
+                visible_to,
                 type_id: 0,
                 group_id: 0,
             })),
@@ -280,6 +295,44 @@ impl TestClient {
         self.recv_message().await
     }
 
+    /// Send a `SettleMarket` message.
+    ///
+    /// # Errors
+    /// Returns an error if sending fails.
+    pub async fn settle_market(
+        &mut self,
+        market_id: i64,
+        settle_price: f64,
+    ) -> anyhow::Result<ServerMessage> {
+        let request_id = self.next_request_id();
+        let msg = ClientMessage {
+            request_id,
+            message: Some(CM::SettleMarket(SettleMarket {
+                market_id,
+                settle_price,
+            })),
+        };
+        self.send_message(msg).await?;
+        self.recv_message().await
+    }
+
+    /// Send an `EditMarket` message.
+    ///
+    /// # Errors
+    /// Returns an error if sending fails.
+    pub async fn edit_market(
+        &mut self,
+        edit: EditMarket,
+    ) -> anyhow::Result<ServerMessage> {
+        let request_id = self.next_request_id();
+        let msg = ClientMessage {
+            request_id,
+            message: Some(CM::EditMarket(edit)),
+        };
+        self.send_message(msg).await?;
+        self.recv_message().await
+    }
+
     /// Send a `CreateOrder` message.
     ///
     /// # Errors