open_chess_server/http/
feed.rs

1//! `GET /api/feed`: every game on the server as server-sent events, for
2//! sites and dashboards that follow it (esports-lab follows it for its
3//! ratings and pricing). No token needed; it shows what the public game
4//! streams show.
5//!
6//! Events:
7//!
8//! * `start`: a game began (the public game description, as the first
9//!   line of `/api/stream/game/{id}`).
10//! * `move`: it moved on (the same description: position, last move,
11//!   ply). While a game is live its moves come
12//!   `CHESS_PUBLIC_DELAY_PLIES` behind, as on the public stream.
13//! * `end`: it finished: the game as `/game/export/{id}` gives it in JSON
14//!   (players, ratings and rating changes, status, winner, moves, clock).
15//!   Its SSE id is when it finished (Unix ms).
16//!
17//! A client that reconnects with `Last-Event-ID` (browsers do it
18//! themselves), or asks with `?since=<Unix ms>`, first gets the `end` of
19//! every game finished since then, oldest first, then the live events.
20//! `?game=<id>` follows one game; `?only=end` sends only finished games.
21//! A comment line every 15 seconds keeps proxies from closing the stream.
22
23use std::collections::{HashMap, HashSet};
24use std::convert::Infallible;
25use std::time::Duration;
26
27use axum::extract::{Query, State};
28use axum::http::HeaderMap;
29use axum::response::sse::{Event, KeepAlive, Sse};
30use axum::response::{IntoResponse, Response};
31use chess_core::Game;
32use futures_util::stream;
33use tokio::sync::broadcast::error::RecvError;
34use tokio::sync::mpsc;
35
36use crate::app::Ctx;
37use crate::db::Note;
38use crate::error::ApiResult;
39use crate::http::{games, streams};
40use crate::metrics::StreamGuard;
41use crate::model::{self, GAME_SELECT, GameRow};
42
43/// Finished games sent per catch-up query.
44const PAGE: i64 = 200;
45
46#[derive(Debug, Clone, Default)]
47struct Filter {
48    game: Option<String>,
49    only_end: bool,
50}
51
52impl Filter {
53    fn wants(&self, id: &str) -> bool {
54        self.game.as_deref().is_none_or(|g| g == id)
55    }
56}
57
58fn end_event(ctx: &Ctx, row: &GameRow) -> Option<Event> {
59    let game = Game::from_moves(row.initial_fen.as_deref(), &row.moves).ok()?;
60    let json = games::json(row, &game, &HashMap::new(), &ctx.cfg);
61    let mut e = Event::default().event("end").json_data(json).ok()?;
62    if let Some(ms) = row.finished_ms {
63        e = e.id(ms.to_string());
64    }
65    Some(e)
66}
67
68fn view_event(kind: &str, ctx: &Ctx, row: &GameRow) -> Option<Event> {
69    Event::default().event(kind).json_data(streams::describe(row, ctx.cfg.public_delay_plies)).ok()
70}
71
72pub async fn feed(
73    State(ctx): State<Ctx>,
74    h: HeaderMap,
75    Query(q): Query<HashMap<String, String>>,
76) -> ApiResult<Response> {
77    let filter = Filter {
78        game: q.get("game").filter(|g| !g.is_empty()).cloned(),
79        only_end: q.get("only").is_some_and(|o| o == "end"),
80    };
81    let since: Option<i64> = h
82        .get("last-event-id")
83        .and_then(|v| v.to_str().ok())
84        .or_else(|| q.get("since").map(String::as_str))
85        .and_then(|s| s.trim().parse().ok());
86    // Subscribed before catching up, so nothing falls between the two.
87    let notes = ctx.db.subscribe();
88    let (tx, rx) = mpsc::channel::<Event>(256);
89    tokio::spawn(run(ctx, filter, since, notes, tx));
90    let events =
91        stream::unfold(rx, |mut rx| async move { rx.recv().await.map(|e| (Ok::<_, Infallible>(e), rx)) });
92    Ok(Sse::new(events).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response())
93}
94
95async fn run(
96    ctx: Ctx,
97    filter: Filter,
98    since: Option<i64>,
99    mut notes: tokio::sync::broadcast::Receiver<Note>,
100    tx: mpsc::Sender<Event>,
101) {
102    let _g = StreamGuard::new(&ctx.metrics, "feed");
103    // Games we've reported finished (an `end` isn't sent twice).
104    let mut ended: HashSet<String> = HashSet::new();
105    // Plies seen of the live games we've reported.
106    let mut seen: HashMap<String, i32> = HashMap::new();
107    if let Some(mut after) = since {
108        loop {
109            let rows = match catch_up(&ctx, &filter, after).await {
110                Ok(r) => r,
111                Err(e) => {
112                    tracing::warn!("feed catch-up: {e:#}");
113                    break;
114                }
115            };
116            let n = rows.len();
117            for row in rows {
118                after = after.max(row.finished_ms.unwrap_or(after));
119                if let Some(e) = end_event(&ctx, &row)
120                    && ended.insert(row.id.clone())
121                    && tx.send(e).await.is_err()
122                {
123                    return;
124                }
125            }
126            if i64::try_from(n).unwrap_or(0) < PAGE {
127                break;
128            }
129            // The next page starts after the last one's time.
130            after += 1;
131        }
132    }
133    loop {
134        let id = match notes.recv().await {
135            Ok(Note::Game(id)) if filter.wants(&id) => id,
136            Ok(_) | Err(RecvError::Lagged(_)) => continue,
137            Err(RecvError::Closed) => return,
138        };
139        let row = match ctx.db.conn().await {
140            Ok(c) => match model::game(&ctx.db, &*c, &id).await {
141                Ok(Some(r)) => r,
142                Ok(None) => continue,
143                Err(e) => {
144                    tracing::warn!("feed: {e:#}");
145                    continue;
146                }
147            },
148            Err(e) => {
149                tracing::warn!("feed: {e:#}");
150                continue;
151            }
152        };
153        let event = if !row.status.is_ongoing() {
154            seen.remove(&row.id);
155            if !ended.insert(row.id.clone()) {
156                continue;
157            }
158            end_event(&ctx, &row)
159        } else if filter.only_end {
160            continue;
161        } else {
162            match seen.insert(row.id.clone(), row.ply) {
163                None => view_event("start", &ctx, &row),
164                Some(ply) if ply != row.ply => view_event("move", &ctx, &row),
165                Some(_) => continue,
166            }
167        };
168        if let Some(e) = event
169            && tx.send(e).await.is_err()
170        {
171            return;
172        }
173        // Keep the bookkeeping small on a long-lived stream.
174        if ended.len() > 50_000 {
175            ended.clear();
176        }
177    }
178}
179
180/// Games finished at or after `after` (Unix ms), oldest first.
181async fn catch_up(ctx: &Ctx, filter: &Filter, after: i64) -> anyhow::Result<Vec<GameRow>> {
182    let c = ctx.db.conn().await?;
183    #[allow(clippy::cast_precision_loss)] // ms since 1970 fit an f64 exactly
184    let at = after as f64 / 1000.0;
185    let rows = c
186        .query(
187            &ctx.db.sql(&format!(
188                "{GAME_SELECT} WHERE g.finished_at >= to_timestamp($1) AND ($2::text IS NULL OR g.id = $2)
189                  ORDER BY g.finished_at, g.id LIMIT {PAGE}"
190            )),
191            &[&at, &filter.game],
192        )
193        .await?;
194    Ok(rows.iter().map(GameRow::from_row).collect())
195}
open_chess_server/http/feed.rs

open_chess_server/http/
feed.rs
