open_chess_server/http/
challenges.rs

1//! Challenge endpoints.
2
3use std::time::Duration;
4
5use axum::Json;
6use axum::extract::{Path, State};
7use axum::response::{IntoResponse, Response};
8use bytes::Bytes;
9use serde_json::json;
10use tokio::sync::mpsc;
11use tracing::warn;
12
13use crate::app::{Ctx, Who};
14use crate::challenges::{self, Request};
15use crate::db::Note;
16use crate::error::{ApiError, ApiResult};
17use crate::http::{form, line, ndjson, ok};
18use crate::metrics::StreamGuard;
19use crate::model;
20
21const PLAY: &[&str] = &["challenge:write", "board:play", "bot:play"];
22
23pub async fn create(
24    State(ctx): State<Ctx>,
25    who: Who,
26    Path(username): Path<String>,
27    body: Bytes,
28) -> ApiResult<Response> {
29    let id = who.need(PLAY)?.clone();
30    let req = Request::parse(&form(&body))?;
31    let (ch, ratings) = challenges::create(&ctx, &id, &username, &req).await?;
32    let json = ch.json(ctx.cfg.public_url(), Some(id.player), ratings);
33    if !req.keep_alive || ch.status != "created" {
34        return Ok(Json(json).into_response());
35    }
36    // keepAliveStream: the challenge lives while this stream is open;
37    // when it's answered, `{"done": "<status>"}` and the stream ends.
38    let slot = ctx.slot(id.player, "challenge")?;
39    let (tx, rx) = mpsc::channel::<Bytes>(16);
40    let _ = tx.send(line(&json)).await;
41    tokio::spawn(async move {
42        keep_alive(ctx, ch.id.clone(), id.player, tx).await;
43        drop(slot);
44    });
45    Ok(ndjson(rx))
46}
47
48async fn keep_alive(ctx: Ctx, id: String, player: i64, tx: mpsc::Sender<Bytes>) {
49    let _g = StreamGuard::new(&ctx.metrics, "challenge");
50    let mut notes = ctx.db.subscribe();
51    let mut refresh = tokio::time::interval(Duration::from_secs(5));
52    let mut keep = tokio::time::interval(Duration::from_secs(6));
53    let done = |status: &str| line(&json!({"done": status}));
54    loop {
55        tokio::select! {
56            n = notes.recv() => {
57                let relevant = match &n {
58                    Ok(Note::User { player: p, event }) => *p == player
59                        && (event.pointer("/challenge/id").and_then(|v| v.as_str()) == Some(id.as_str())
60                            || event.pointer("/game/id").and_then(|v| v.as_str()) == Some(id.as_str())),
61                    Ok(Note::Resync) | Err(_) => true,
62                    _ => false,
63                };
64                if relevant {
65                    match challenges::status(&ctx, &id).await {
66                        Ok(Some(s)) if s != "created" => {
67                            let _ = tx.send(done(&s)).await;
68                            return;
69                        }
70                        Ok(_) => {}
71                        Err(e) => warn!("challenge stream: {e:#}"),
72                    }
73                }
74            }
75            _ = refresh.tick() => {
76                if let Err(e) = challenges::keep_alive(&ctx, &id).await {
77                    warn!("keeping challenge {id} alive: {e:#}");
78                }
79            }
80            _ = keep.tick() => {
81                if tx.send(Bytes::from_static(b"\n")).await.is_err() {
82                    break;
83                }
84            }
85        }
86    }
87    // The challenger went away: the challenge goes with them.
88    let who = crate::auth::Identity {
89        player,
90        token_id: 0,
91        scopes: Vec::new(),
92        admin: false,
93        manager: false,
94        expires_ms: 0,
95    };
96    let _ = challenges::cancel(&ctx, &who, &id).await;
97}
98
99pub async fn list(State(ctx): State<Ctx>, who: Who) -> ApiResult<Response> {
100    let id = who.need(&["challenge:read", "challenge:write", "board:play", "bot:play"])?;
101    let (incoming, outgoing) = challenges::list(&ctx, id.player).await?;
102    Ok(Json(chess_proto::ChallengeList { incoming, outgoing }).into_response())
103}
104
105pub async fn show(State(ctx): State<Ctx>, who: Who, Path(id): Path<String>) -> ApiResult<Response> {
106    let me = who.need(&["challenge:read", "challenge:write", "board:play", "bot:play"])?;
107    let c = ctx.db.conn().await?;
108    let ch =
109        model::challenge(&ctx.db, &*c, &id, false).await?.ok_or_else(|| ApiError::not_found("challenge"))?;
110    if ch.challenger.player_id != me.player && ch.dest.player_id != me.player {
111        return Err(ApiError::not_found("challenge"));
112    }
113    let ratings = model::challenge_ratings(&ctx.db, &*c, &ch).await?;
114    Ok(Json(ch.json(ctx.cfg.public_url(), Some(me.player), ratings)).into_response())
115}
116
117pub async fn accept(State(ctx): State<Ctx>, who: Who, Path(id): Path<String>) -> ApiResult<Response> {
118    let me = who.need(PLAY)?;
119    challenges::accept(&ctx, me, &id).await?;
120    Ok(ok())
121}
122
123pub async fn decline(
124    State(ctx): State<Ctx>,
125    who: Who,
126    Path(id): Path<String>,
127    body: Bytes,
128) -> ApiResult<Response> {
129    let me = who.need(PLAY)?;
130    let f = form(&body);
131    challenges::decline(&ctx, me, &id, f.get("reason").map(String::as_str)).await?;
132    Ok(ok())
133}
134
135pub async fn cancel(State(ctx): State<Ctx>, who: Who, Path(id): Path<String>) -> ApiResult<Response> {
136    let me = who.need(PLAY)?;
137    challenges::cancel(&ctx, me, &id).await?;
138    Ok(ok())
139}
open_chess_server/http/challenges.rs

open_chess_server/http/
challenges.rs
