WebSockets

Popular concerts sell out in minutes, and fans keep refreshing the page to see which seats are left. Let's show that live instead. The page opens a WebSocket, and the box office tells it whenever seats sell.

Margaret's WebSockets speak a small JSON protocol: the page sends requests, and the box office answers each one with a single response or a stream of them. Add serde_json to your dependencies.

Taken seats

A seat map needs to know which seats are taken. The unique constraint on [concert, seat] lists a concert's tickets by seat, so the repository can read them all, 500 at a time, as a stream:

#[singleton]
pub struct TicketRepository {
    database: Arc<Database>,
}

impl TicketRepository {
    #[constructor]
    pub fn create(database: Arc<Database>) -> anyhow::Result<Self> {
        Ok(Self { database })
    }

    pub async fn taken_seats(&self, concert: &Concert) -> Result<Vec<i32>, ActiveRecordError> {
        Ticket::query()
            .concert
            .eq(Key::of(concert))
            .seat
            .ascending()
            .stream::<500, _>(self.database.as_ref())
            .map_ok(|ticket| ticket.seat)
            .try_collect()
            .await
    }
}

stream reads in batches of the size you pick, and yields records one by one, so even a stadium never ends up in a single read.

Sessions

Every connection gets a session. It lives as long as the connection, so it's the place for anything that belongs to one visitor:

use margaret::framework::macros::build_for_session;
use margaret::framework::macros::websocket_session;
use tokio::sync::watch;

use crate::concerts::concert::Concert;

#[websocket_session(path = "/concerts/{concert}/live", server = "public")]
pub struct ConcertSession {
    pub concert: Concert,
    pub visible: watch::Sender<bool>,
}

impl ConcertSession {
    #[build_for_session]
    pub fn build(#[route_parameter(from = "concert")] concert: Concert) -> anyhow::Result<Self> {
        Ok(Self {
            concert,
            visible: watch::Sender::new(true),
        })
    }
}

#[build_for_session] runs before the connection opens, and it binds route parameters just like a route does. An unknown concert gets a 404, and no WebSocket at all.

Messages

Messages are structs with #[websocket_message]. The page sends a request to start watching, and the box office streams responses back:

use margaret::framework::macros::websocket_message;
use margaret::framework::websocket::web_socket_response::WebSocketResponse;
use serde::Deserialize;
use validator::Validate;

#[websocket_message(request, method = "watch_seats", response = WebSocketResponse::Stream)]
#[derive(Deserialize, Validate)]
pub struct WatchSeats {}

use margaret::framework::macros::websocket_message;
use serde::Serialize;

#[websocket_message(response, method = "seats_taken")]
#[derive(Serialize)]
pub struct SeatsTaken {
    pub seats: Vec<i32>,
    pub remaining: usize,
}

Streaming responses

A handler is a singleton that implements RespondsToWebSocketMessage. Its Session and Message types tie it to a session and a message. This one sends the taken seats, then checks again every two seconds, and sends an update whenever they change:

#[async_trait]
impl RespondsToWebSocketMessage for WatchSeatsHandler {
    type Message = WatchSeats;
    type Session = ConcertSession;

    async fn process(
        &self,
        cancellation_token: CancellationToken,
        session: Arc<ConcertSession>,
        message: StreamingRequestEnvelope<WatchSeats>,
        socket: WebSocket,
    ) -> anyhow::Result<()> {
        let capacity = usize::try_from(session.concert.capacity)?;
        let mut visible = session.visible.subscribe();
        let mut last_sent = None;

        loop {
            if *visible.borrow_and_update() {
                let seats = self.tickets.taken_seats(&session.concert).await?;
                let remaining = capacity.saturating_sub(seats.len());

                if remaining == 0 {
                    socket
                        .send(message.fin(SeatsTaken { seats, remaining }))
                        .await?;

                    return Ok(());
                }

                if last_sent.as_ref() != Some(&seats) {
                    socket
                        .send(message.chunk(SeatsTaken {
                            seats: seats.clone(),
                            remaining,
                        }))
                        .await?;
                    last_sent = Some(seats);
                }
            }

            tokio::select! {
                () = cancellation_token.cancelled() => return Ok(()),
                changed = visible.changed() => changed?,
                () = tokio::time::sleep(SEAT_REFRESH_INTERVAL), if *visible.borrow() => {}
            }
        }
    }
}

Each chunk is one update. fin ends the stream, here when the concert sells out. Every instance of the box office reads the same database, so fans see seats sold by any instance. Other requests on the same connection keep working while a stream runs.

Single responses

Before buying, the seat picker asks for the total price. That request gets exactly one answer, so it's declared with WebSocketResponse::Single:

#[websocket_message(request, method = "quote", response = WebSocketResponse::Single)]
#[derive(Deserialize, Validate)]
pub struct Quote {
    #[validate(range(min = 1, max = 8))]
    pub seats: u32,
}

#[async_trait]
impl RespondsToWebSocketMessage for QuoteHandler {
    type Message = Quote;
    type Session = ConcertSession;

    async fn process(
        &self,
        _cancellation_token: CancellationToken,
        session: Arc<ConcertSession>,
        message: RequestEnvelope<Quote>,
        socket: WebSocket,
    ) -> anyhow::Result<()> {
        let total = session.concert.price * Decimal::from(message.message().seats);

        socket.send(message.response(Quoted { total })).await?;

        Ok(())
    }
}

Notifications

Notifications are messages that expect no answer. When a fan switches to another tab, the seat picker sends visibility, and the stream above pauses instead of reading the database for nobody:

use margaret::framework::macros::websocket_message;
use serde::Deserialize;
use validator::Validate;

#[websocket_message(notification, method = "visibility")]
#[derive(Deserialize, Validate)]
pub struct Visibility {
    pub visible: bool,
}

#[async_trait]
impl RespondsToWebSocketNotification for VisibilityHandler {
    type Message = Visibility;
    type Session = ConcertSession;

    async fn process(
        &self,
        _cancellation_token: CancellationToken,
        session: Arc<ConcertSession>,
        message: NotificationEnvelope<Visibility>,
        _socket: WebSocket,
    ) -> anyhow::Result<()> {
        session.visible.send_replace(message.message().visible);

        Ok(())
    }
}

The protocol

Requests carry an id, and notifications don't:

{"id": 1, "method": "watch_seats", "params": {}}
{"id": 2, "method": "quote", "params": {"seats": 2}}
{"method": "visibility", "params": {"visible": false}}

Every response carries the request's id. done stays false until the last response:

{"id":2,"done":true,"method":"quoted","result":{"total":"65.00"}}
{"id":1,"done":false,"method":"seats_taken","result":{"seats":[1,2,7],"remaining":397}}

Errors look like this:

{"error":{"code":"unknown_method","details":null,"message":"no handler is registered for the request method 'nope'"},"id":3}
{"error":{"code":"invalid_params","details":null,"message":"the message parameters failed validation"},"id":4}

What the build checks

A message that nothing handles, or one handled twice in the same session, is a build error.