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

open_chess_server/http/
feed.rs
