open_chess_server/
db.rs

1//! Postgres: the connection pool, this server's schema, and the LISTEN
2//! connection that fans changes out to every replica.
3//!
4//! SQL is written with `chess.` table prefixes; [`Db::sql`] swaps in the
5//! configured schema name. The server keeps its own players, API tokens
6//! and sessions there; a host site's (`site.`, `CHESS_SITE_SCHEMA`) are read
7//! once, to import the players who already used chess there. Explicit names rather than a
8//! `search_path` because Neon's pooler (transaction pooling) doesn't keep
9//! session settings.
10
11use std::fmt::Write as _;
12use std::str::FromStr;
13use std::sync::Arc;
14use std::time::Duration;
15
16use anyhow::{Context, Result, bail};
17use deadpool::managed::{self, Metrics, RecycleError, RecycleResult};
18use futures_util::StreamExt;
19use tokio::sync::broadcast;
20use tracing::{info, warn};
21
22use crate::config::Config;
23
24pub fn quote_ident(s: &str) -> String {
25    format!("\"{}\"", s.replace('"', "\"\""))
26}
27
28fn tls() -> Result<tokio_postgres_rustls::MakeRustlsConnect> {
29    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
30    let mut roots = rustls::RootCertStore::empty();
31    let native = rustls_native_certs::load_native_certs();
32    let (added, _) = roots.add_parsable_certificates(native.certs);
33    if added == 0 {
34        bail!("no trusted root certificates found for TLS");
35    }
36    let cfg = rustls::ClientConfig::builder().with_root_certificates(roots).with_no_client_auth();
37    Ok(tokio_postgres_rustls::MakeRustlsConnect::new(cfg))
38}
39
40/// Neon's pooled hostname (`ep-x-123-pooler.region…`) → the direct one,
41/// which LISTEN needs (transaction poolers don't deliver notifications).
42pub fn direct_host(host: &str) -> Option<String> {
43    let (first, rest) = host.split_once('.')?;
44    first.strip_suffix("-pooler").map(|ep| format!("{ep}.{rest}"))
45}
46
47pub struct Manager {
48    pg: tokio_postgres::Config,
49    tls: tokio_postgres_rustls::MakeRustlsConnect,
50}
51
52impl managed::Manager for Manager {
53    type Type = tokio_postgres::Client;
54    type Error = tokio_postgres::Error;
55
56    async fn create(&self) -> Result<tokio_postgres::Client, tokio_postgres::Error> {
57        let (client, conn) = self.pg.connect(self.tls.clone()).await?;
58        tokio::spawn(async move {
59            if let Err(e) = conn.await {
60                warn!("postgres connection: {e}");
61            }
62        });
63        Ok(client)
64    }
65
66    async fn recycle(
67        &self,
68        client: &mut tokio_postgres::Client,
69        _: &Metrics,
70    ) -> RecycleResult<tokio_postgres::Error> {
71        if client.is_closed() {
72            return Err(RecycleError::message("connection closed"));
73        }
74        Ok(())
75    }
76}
77
78pub type Pool = managed::Pool<Manager>;
79pub type Conn = managed::Object<Manager>;
80
81/// A change other replicas (and local streams) need to hear about.
82#[derive(Debug, Clone)]
83pub enum Note {
84    /// A game's state changed: reload it.
85    Game(String),
86    /// A chat line in a game.
87    Chat { game: String, line: serde_json::Value },
88    /// An event for one user's event stream (already in its final JSON).
89    User { player: i64, event: serde_json::Value },
90    /// The LISTEN connection dropped and came back: notes may have been
91    /// missed, so every stream reloads.
92    Resync,
93}
94
95#[derive(Clone)]
96pub struct Db {
97    pub pool: Pool,
98    schema: Arc<str>,
99    site: Arc<str>,
100    channel: Arc<str>,
101    notes: broadcast::Sender<Note>,
102}
103
104impl Db {
105    pub fn connect(cfg: &Config) -> Result<Db> {
106        let pg = tokio_postgres::Config::from_str(&cfg.database_url)
107            .context("DATABASE_URL isn't a postgres:// URL")?;
108        let pool = Pool::builder(Manager { pg, tls: tls()? })
109            .max_size(cfg.pool_size.max(2))
110            .wait_timeout(Some(Duration::from_secs(10)))
111            .create_timeout(Some(Duration::from_secs(15)))
112            .runtime(deadpool::Runtime::Tokio1)
113            .build()
114            .context("building the database pool")?;
115        let (notes, _) = broadcast::channel(4096);
116        Ok(Db {
117            pool,
118            schema: cfg.schema.as_str().into(),
119            site: cfg.site_schema.as_str().into(),
120            channel: format!("{}_events", cfg.schema).into(),
121            notes,
122        })
123    }
124
125    /// Swap `chess.` and `site.` table prefixes for the configured schemas
126    /// (only where a prefix starts a word, so prose in comments is safe).
127    pub fn sql(&self, q: &str) -> String {
128        let c = format!("{}.", quote_ident(&self.schema));
129        let s = format!("{}.", quote_ident(&self.site));
130        prefix_replace(&prefix_replace(q, "chess.", &c), "site.", &s)
131    }
132
133    pub async fn conn(&self) -> Result<Conn> {
134        self.pool.get().await.map_err(|e| anyhow::anyhow!("database unavailable: {e}"))
135    }
136
137    pub fn channel(&self) -> &str {
138        &self.channel
139    }
140
141    pub fn subscribe(&self) -> broadcast::Receiver<Note> {
142        self.notes.subscribe()
143    }
144
145    pub fn encode(n: &Note) -> String {
146        let v = match n {
147            Note::Game(id) => serde_json::json!({"t": "g", "g": id}),
148            Note::Chat { game, line } => serde_json::json!({"t": "c", "g": game, "l": line}),
149            Note::User { player, event } => serde_json::json!({"t": "u", "u": player, "e": event}),
150            Note::Resync => serde_json::json!({"t": "r"}),
151        };
152        v.to_string()
153    }
154
155    pub fn decode(s: &str) -> Option<Note> {
156        let v: serde_json::Value = serde_json::from_str(s).ok()?;
157        match v.get("t")?.as_str()? {
158            "g" => Some(Note::Game(v.get("g")?.as_str()?.to_owned())),
159            "c" => Some(Note::Chat { game: v.get("g")?.as_str()?.to_owned(), line: v.get("l")?.clone() }),
160            "u" => Some(Note::User { player: v.get("u")?.as_i64()?, event: v.get("e")?.clone() }),
161            "r" => Some(Note::Resync),
162            _ => None,
163        }
164    }
165
166    /// Run migrations: this server's schema and tables, then (once) the
167    /// import of a host site's chess players ([`Db::import_site`]).
168    pub async fn migrate(&self) -> Result<()> {
169        let mut c = self.conn().await?;
170        let tx = c.transaction().await?;
171        tx.batch_execute(&format!(
172            "SELECT pg_advisory_xact_lock(hashtext('open-chess-migrate'));
173             CREATE SCHEMA IF NOT EXISTS {s};
174             CREATE TABLE IF NOT EXISTS {s}.schema_version (version integer NOT NULL);",
175            s = quote_ident(&self.schema)
176        ))
177        .await?;
178        // Players, tokens and sessions first: the migrations refer to them.
179        tx.batch_execute(&self.sql(IDENTITY)).await.context("chess identity tables")?;
180        let current: i32 = tx
181            .query_opt(&self.sql("SELECT version FROM chess.schema_version"), &[])
182            .await?
183            .map_or(0, |r| r.get(0));
184        for (i, m) in MIGRATIONS.iter().enumerate() {
185            let v = i32::try_from(i + 1).unwrap_or(i32::MAX);
186            if v > current {
187                tx.batch_execute(&self.sql(m))
188                    .await
189                    .with_context(|| format!("chess schema migration {v}"))?;
190                info!(version = v, "chess schema migrated");
191            }
192        }
193        let latest = i32::try_from(MIGRATIONS.len()).unwrap_or(i32::MAX);
194        if current == 0 {
195            tx.execute(&self.sql("INSERT INTO chess.schema_version VALUES ($1)"), &[&latest]).await?;
196        } else if latest > current {
197            tx.execute(&self.sql("UPDATE chess.schema_version SET version = $1"), &[&latest]).await?;
198        }
199        self.import_site(&tx).await.context("importing the site's chess players")?;
200        tx.commit().await?;
201        Ok(())
202    }
203
204    /// Once: copy the host site's players who use chess (their roles stay
205    /// the site's; the site vouches for their emails) (they have a chess
206    /// account, or are named in any of our tables, or hold a token with a
207    /// chess scope) with their ids, and their unexpired chess tokens, into
208    /// our own tables, linked to the site by `esports_id`; then point our
209    /// tables' references at our players. A site without those tables (a
210    /// stand-alone server) has nothing to import.
211    async fn import_site(&self, tx: &tokio_postgres::Transaction<'_>) -> Result<()> {
212        if *self.site == *self.schema {
213            return Ok(());
214        }
215        let has = |t: &'static str| {
216            let site = self.site.clone();
217            async move {
218                tx.query_one(
219                    "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2)",
220                    &[&&*site, &t],
221                )
222                .await
223                .map(|r| r.get::<_, bool>(0))
224            }
225        };
226        if !(has("players").await? && has("api_tokens").await?) {
227            return Ok(());
228        }
229        let done = tx
230            .query_opt(&self.sql("SELECT 1 FROM chess.imports WHERE source = $1"), &[&&*self.site])
231            .await?
232            .is_some();
233        if done {
234            return Ok(());
235        }
236        let site_players = format!("{}.players", quote_ident(&self.site));
237        // Our foreign keys still pointing at the site's players.
238        let fks = tx
239            .query(
240                "SELECT c.conname, c.conrelid::regclass::text, a.attname, pg_get_constraintdef(c.oid)
241                   FROM pg_constraint c
242                   JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = c.conkey[1]
243                  WHERE c.contype = 'f' AND c.confrelid = to_regclass($1)
244                    AND c.connamespace = (SELECT oid FROM pg_namespace WHERE nspname = $2)",
245                &[&site_players, &&*self.schema],
246            )
247            .await?;
248        let scopes: Vec<String> = crate::auth::SCOPES.iter().map(|(s, _)| (*s).to_owned()).collect();
249        let mut named = String::from("SELECT player_id FROM site.api_tokens WHERE scopes && $1::text[]");
250        for f in &fks {
251            let (table, column): (String, String) = (f.get(1), f.get(2));
252            let _ = write!(named, " UNION SELECT {} FROM {table}", quote_ident(&column));
253        }
254        let players = tx
255            .execute(
256                &self.sql(&format!(
257                    "INSERT INTO chess.players (id, name, email, email_verified_at, key_hash, disabled_at, esports_id)
258                     SELECT p.id, p.name, p.email, CASE WHEN p.email IS NULL THEN NULL ELSE now() END, p.key_hash,
259                            p.disabled_at, p.id::text
260                       FROM site.players p WHERE p.id IN ({named})
261                     ON CONFLICT (id) DO NOTHING"
262                )),
263                &[&scopes],
264            )
265            .await?;
266        let tokens = tx
267            .execute(
268                &self.sql(
269                    "INSERT INTO chess.api_tokens (player_id, name, token_hash, prefix, scopes, created_at,
270                                                   expires_at, last_used_at, revoked_at, revoked_by)
271                     SELECT t.player_id, t.name, t.token_hash, t.prefix,
272                            ARRAY(SELECT unnest(t.scopes) INTERSECT SELECT unnest($1::text[])),
273                            t.created_at, t.expires_at, t.last_used_at, t.revoked_at, t.revoked_by
274                       FROM site.api_tokens t
275                      WHERE t.player_id IN (SELECT id FROM chess.players) AND t.scopes && $1::text[]
276                        AND t.revoked_at IS NULL AND t.expires_at > now()
277                     ON CONFLICT (token_hash) DO NOTHING",
278                ),
279                &[&scopes],
280            )
281            .await?;
282        tx.execute(
283            &self.sql(
284                "SELECT setval(pg_get_serial_sequence($1, 'id'), GREATEST((SELECT max(id) FROM chess.players), 1))",
285            ),
286            &[&self.sql("chess.players")],
287        )
288        .await?;
289        let ours = self.sql("chess.players");
290        for f in &fks {
291            let (name, table, column, def): (String, String, String, String) =
292                (f.get(0), f.get(1), f.get(2), f.get(3));
293            let tail = def.split_once(".players(id)").map_or("", |(_, t)| t);
294            tx.batch_execute(&format!(
295                "ALTER TABLE {table} DROP CONSTRAINT {n};
296                 ALTER TABLE {table} ADD CONSTRAINT {n} FOREIGN KEY ({c}) REFERENCES {ours}(id){tail};",
297                n = quote_ident(&name),
298                c = quote_ident(&column),
299            ))
300            .await?;
301        }
302        tx.execute(&self.sql("INSERT INTO chess.imports (source) VALUES ($1)"), &[&&*self.site]).await?;
303        info!(
304            players,
305            tokens,
306            references = fks.len(),
307            site = &*self.site,
308            "imported the site's chess players"
309        );
310        Ok(())
311    }
312
313    /// Keep a LISTEN connection open and forward notifications into the
314    /// local broadcast. Reconnects with backoff; after a reconnect every
315    /// stream is told to resync.
316    pub async fn listen(self, cfg: Config, mut stop: tokio::sync::watch::Receiver<bool>) {
317        let mut backoff = Duration::from_millis(500);
318        let mut first = true;
319        loop {
320            let started = std::time::Instant::now();
321            match self.listen_once(&cfg, first).await {
322                Ok(()) => {}
323                Err(e) => warn!("chess LISTEN connection: {e:#}"),
324            }
325            first = false;
326            if started.elapsed() > Duration::from_secs(30) {
327                backoff = Duration::from_millis(500);
328            }
329            tokio::select! {
330                () = tokio::time::sleep(backoff) => {}
331                () = crate::stopped(&mut stop) => return,
332            }
333            backoff = (backoff * 2).min(Duration::from_secs(10));
334        }
335    }
336
337    async fn listen_once(&self, cfg: &Config, first: bool) -> Result<()> {
338        let url = match &cfg.listen_database_url {
339            Some(u) => u.clone(),
340            None => listen_url(&cfg.database_url),
341        };
342        let pg = tokio_postgres::Config::from_str(&url).context("the LISTEN database URL")?;
343        let (client, mut conn) = pg.connect(tls()?).await?;
344        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
345        let driver = tokio::spawn(async move {
346            let mut s = futures_util::stream::poll_fn(move |cx| conn.poll_message(cx));
347            while let Some(m) = s.next().await {
348                match m {
349                    Ok(tokio_postgres::AsyncMessage::Notification(n)) => {
350                        if tx.send(n.payload().to_owned()).is_err() {
351                            break;
352                        }
353                    }
354                    Ok(_) => {}
355                    Err(e) => {
356                        warn!("chess LISTEN: {e}");
357                        break;
358                    }
359                }
360            }
361        });
362        client.batch_execute(&format!("LISTEN {}", quote_ident(&self.channel))).await?;
363        info!(channel = &*self.channel, "listening for chess changes");
364        if !first {
365            let _ = self.notes.send(Note::Resync);
366        }
367        let mut ping = tokio::time::interval(Duration::from_secs(30));
368        loop {
369            tokio::select! {
370                m = rx.recv() => match m {
371                    Some(p) => {
372                        if let Some(n) = Db::decode(&p) {
373                            let _ = self.notes.send(n);
374                        }
375                    }
376                    None => break,
377                },
378                _ = ping.tick() => {
379                    if client.simple_query("SELECT 1").await.is_err() {
380                        break;
381                    }
382                }
383            }
384        }
385        driver.abort();
386        bail!("the LISTEN connection closed")
387    }
388}
389
390/// The URL for LISTEN: a Neon `-pooler` host swapped for the direct one
391/// (the host is replaced, not added: tokio-postgres would try a second
392/// host only if the first failed, and the pooler doesn't fail, it just
393/// never delivers notifications). Anything else is used as given.
394pub fn listen_url(database_url: &str) -> String {
395    let Ok(mut u) = url::Url::parse(database_url) else { return database_url.to_owned() };
396    let Some(direct) = u.host_str().and_then(direct_host) else { return database_url.to_owned() };
397    if u.set_host(Some(&direct)).is_err() {
398        return database_url.to_owned();
399    }
400    u.into()
401}
402
403/// Replace `from` with `to` wherever it starts a word and is followed by a
404/// letter (a table name).
405fn prefix_replace(q: &str, from: &str, to: &str) -> String {
406    let mut out = String::with_capacity(q.len() + 32);
407    let mut rest = q;
408    let mut prev: Option<char> = None;
409    while let Some(i) = rest.find(from) {
410        let before = rest[..i].chars().next_back().or(prev);
411        let after = rest[i + from.len()..].chars().next();
412        out.push_str(&rest[..i]);
413        let starts_word = before.is_none_or(|c| !(c.is_alphanumeric() || c == '_' || c == '"' || c == '.'));
414        if starts_word && after.is_some_and(|c| c.is_ascii_alphabetic()) {
415            out.push_str(to);
416        } else {
417            out.push_str(from);
418        }
419        prev = from.chars().next_back();
420        rest = &rest[i + from.len()..];
421    }
422    out.push_str(rest);
423    out
424}
425
426/// Send `note` from inside a transaction (delivered to every replica on
427/// commit). Payloads over Postgres's 8000-byte limit are trimmed to a
428/// reload instruction.
429pub async fn notify(tx: &impl tokio_postgres::GenericClient, db: &Db, note: &Note) -> Result<()> {
430    let mut payload = Db::encode(note);
431    if payload.len() > 7900 {
432        payload = match note {
433            Note::Game(id) | Note::Chat { game: id, .. } => Db::encode(&Note::Game(id.clone())),
434            _ => Db::encode(&Note::Resync),
435        };
436    }
437    // Postgres delivers on commit, to every replica listening.
438    tx.execute("SELECT pg_notify($1, $2)", &[&db.channel(), &payload]).await?;
439    Ok(())
440}
441
442/// The server's own players, API tokens and sessions (created before the
443/// migrations, which refer to them; idempotent). `esports_id` links a
444/// player to the host site's (signed in with it, or imported from it).
445const IDENTITY: &str = r"
446CREATE TABLE IF NOT EXISTS chess.players (
447    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
448    name text NOT NULL,
449    email text,
450    key_hash text,
451    password_hash text,
452    role text NOT NULL DEFAULT 'player',
453    roles text[] NOT NULL DEFAULT '{}',
454    esports_id text UNIQUE,
455    created_at timestamptz NOT NULL DEFAULT now(),
456    disabled_at timestamptz);
457CREATE TABLE IF NOT EXISTS chess.api_tokens (
458    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
459    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
460    name text NOT NULL,
461    token_hash text NOT NULL UNIQUE,
462    prefix text NOT NULL,
463    scopes text[] NOT NULL,
464    created_at timestamptz NOT NULL DEFAULT now(),
465    expires_at timestamptz NOT NULL,
466    last_used_at timestamptz,
467    revoked_at timestamptz,
468    revoked_by text);
469CREATE INDEX IF NOT EXISTS api_tokens_player_idx ON chess.api_tokens (player_id);
470CREATE TABLE IF NOT EXISTS chess.sessions (
471    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
472    token_hash text NOT NULL UNIQUE,
473    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
474    created_at timestamptz NOT NULL DEFAULT now(),
475    expires_at timestamptz NOT NULL,
476    revoked_at timestamptz);
477CREATE TABLE IF NOT EXISTS chess.imports (
478    source text PRIMARY KEY,
479    at timestamptz NOT NULL DEFAULT now());
480";
481
482const MIGRATIONS: &[&str] = &[
483    r"
484-- One row per site player who has used the chess server. `username` is
485-- their Lichess-style handle here; `kind` is human or bot (bots play
486-- through the Bot API and are never fair-play screened); `marked_at` is a
487-- confirmed fair-play violation (shown as tosViolation).
488CREATE TABLE chess.accounts (
489    player_id bigint PRIMARY KEY REFERENCES chess.players ON DELETE CASCADE,
490    username text NOT NULL,
491    kind text NOT NULL DEFAULT 'human' CHECK (kind IN ('human', 'bot')),
492    created_at timestamptz NOT NULL DEFAULT now(),
493    seen_at timestamptz,
494    marked_at timestamptz,
495    marked_by bigint,
496    CHECK (username ~ '^[A-Za-z0-9][A-Za-z0-9_-]{1,29}$'));
497CREATE UNIQUE INDEX accounts_username_idx ON chess.accounts (lower(username));
498
499-- Glicko-2 per rating pool (bullet, blitz, …).
500CREATE TABLE chess.ratings (
501    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
502    perf text NOT NULL,
503    rating double precision NOT NULL,
504    deviation double precision NOT NULL,
505    volatility double precision NOT NULL,
506    games integer NOT NULL DEFAULT 0,
507    prog integer NOT NULL DEFAULT 0,
508    updated_at timestamptz NOT NULL DEFAULT now(),
509    PRIMARY KEY (player_id, perf));
510
511CREATE TABLE chess.challenges (
512    id text PRIMARY KEY,
513    challenger_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
514    dest_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
515    status text NOT NULL DEFAULT 'created'
516        CHECK (status IN ('created', 'offline', 'canceled', 'declined', 'accepted')),
517    variant text NOT NULL DEFAULT 'standard' CHECK (variant IN ('standard', 'fromPosition')),
518    initial_fen text,
519    rated boolean NOT NULL DEFAULT false,
520    clock_limit integer,
521    clock_increment integer,
522    color text NOT NULL DEFAULT 'random' CHECK (color IN ('white', 'black', 'random')),
523    final_color text NOT NULL CHECK (final_color IN ('white', 'black')),
524    rules text[] NOT NULL DEFAULT '{}',
525    decline_reason text,
526    created_at timestamptz NOT NULL DEFAULT now(),
527    expires_at timestamptz NOT NULL);
528CREATE INDEX challenges_dest_idx ON chess.challenges (dest_id) WHERE status = 'created';
529CREATE INDEX challenges_challenger_idx ON chess.challenges (challenger_id) WHERE status = 'created';
530CREATE INDEX challenges_expiry_idx ON chess.challenges (expires_at) WHERE status = 'created';
531
532-- Games. `moves` is UCI (standard castling), `turn_started_at` is when the
533-- side to move's clock started (the database clock is the only clock, so
534-- every replica agrees). move_ms / clock_ms: time used and time left per
535-- ply, for exports and fair-play timing.
536CREATE TABLE chess.games (
537    id text PRIMARY KEY,
538    white_id bigint NOT NULL REFERENCES chess.accounts,
539    black_id bigint NOT NULL REFERENCES chess.accounts,
540    variant text NOT NULL DEFAULT 'standard',
541    initial_fen text,
542    rated boolean NOT NULL,
543    source text NOT NULL DEFAULT 'friend',
544    speed text NOT NULL,
545    rules text[] NOT NULL DEFAULT '{}',
546    clock_initial_ms bigint,
547    clock_increment_ms bigint,
548    status smallint NOT NULL DEFAULT 20,
549    winner text CHECK (winner IN ('white', 'black')),
550    moves text NOT NULL DEFAULT '',
551    ply integer NOT NULL DEFAULT 0,
552    fen text NOT NULL,
553    white_ms bigint,
554    black_ms bigint,
555    move_ms bigint[] NOT NULL DEFAULT '{}',
556    clock_ms bigint[] NOT NULL DEFAULT '{}',
557    draw_offer text CHECK (draw_offer IN ('white', 'black')),
558    takeback_offer text CHECK (takeback_offer IN ('white', 'black')),
559    white_rating integer NOT NULL,
560    black_rating integer NOT NULL,
561    white_provisional boolean NOT NULL DEFAULT false,
562    black_provisional boolean NOT NULL DEFAULT false,
563    white_rating_diff integer,
564    black_rating_diff integer,
565    created_at timestamptz NOT NULL DEFAULT now(),
566    turn_started_at timestamptz NOT NULL DEFAULT now(),
567    last_move_at timestamptz,
568    finished_at timestamptz,
569    analysis text NOT NULL DEFAULT 'none'
570        CHECK (analysis IN ('none', 'pending', 'running', 'done', 'failed', 'skipped')),
571    analysis_lease_until timestamptz,
572    CHECK (white_id <> black_id));
573CREATE INDEX games_ongoing_idx ON chess.games (turn_started_at) WHERE status IN (10, 20);
574CREATE INDEX games_white_idx ON chess.games (white_id, created_at DESC);
575CREATE INDEX games_black_idx ON chess.games (black_id, created_at DESC);
576CREATE INDEX games_analysis_idx ON chess.games (finished_at) WHERE analysis IN ('pending', 'running');
577
578CREATE TABLE chess.chat (
579    id bigserial PRIMARY KEY,
580    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
581    room text NOT NULL CHECK (room IN ('player', 'spectator')),
582    player_id bigint,
583    username text NOT NULL,
584    text text NOT NULL CHECK (length(text) <= 140),
585    at timestamptz NOT NULL DEFAULT now());
586CREATE INDEX chat_game_idx ON chess.chat (game_id, id);
587
588-- Who has a game stream open (heartbeats; nothing here needs to survive
589-- a restart, hence unlogged).
590CREATE UNLOGGED TABLE chess.presence (
591    game_id text NOT NULL,
592    player_id bigint NOT NULL,
593    seen_at timestamptz NOT NULL DEFAULT now(),
594    PRIMARY KEY (game_id, player_id));
595
596-- OAuth authorization codes (PKCE), single use, two minutes.
597CREATE TABLE chess.oauth_codes (
598    code_hash text PRIMARY KEY,
599    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
600    client_id text NOT NULL,
601    redirect_uri text NOT NULL,
602    code_challenge text NOT NULL,
603    scopes text[] NOT NULL,
604    expires_at timestamptz NOT NULL,
605    used_at timestamptz);
606
607-- Fair play: one report per screened side of a game, and reviewers'
608-- verdicts.
609CREATE TABLE chess.fairplay_reports (
610    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
611    side text NOT NULL CHECK (side IN ('white', 'black')),
612    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
613    score real NOT NULL,
614    flagged boolean NOT NULL,
615    moves integer NOT NULL,
616    accuracy real NOT NULL,
617    acpl real NOT NULL,
618    top1 real NOT NULL,
619    top3 real NOT NULL,
620    time_cv real,
621    time_corr real,
622    reasons text[] NOT NULL DEFAULT '{}',
623    engine text NOT NULL,
624    depth integer NOT NULL,
625    evals jsonb NOT NULL,
626    created_at timestamptz NOT NULL DEFAULT now(),
627    PRIMARY KEY (game_id, side));
628CREATE INDEX fairplay_flagged_idx ON chess.fairplay_reports (created_at DESC) WHERE flagged;
629CREATE INDEX fairplay_player_idx ON chess.fairplay_reports (player_id, created_at DESC);
630CREATE TABLE chess.fairplay_reviews (
631    id bigserial PRIMARY KEY,
632    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
633    game_id text REFERENCES chess.games ON DELETE SET NULL,
634    verdict text NOT NULL CHECK (verdict IN ('clean', 'engine', 'unclear')),
635    note text,
636    reviewer_id bigint NOT NULL,
637    at timestamptz NOT NULL DEFAULT now());
638",
639    r"
640-- 2. The position dictionary, assistance declarations, and the training-app
641-- fingerprint on fair-play reports.
642--
643-- positions: one row per position (FEN without the move counters), the
644-- deepest engine evaluation stored for it, filled by the fair-play worker
645-- as it screens games. A lookup is a key read.
646CREATE TABLE chess.positions (
647    key text PRIMARY KEY,
648    best text,
649    score_cp integer NOT NULL,
650    depth integer NOT NULL,
651    engine text NOT NULL,
652    seen integer NOT NULL DEFAULT 1,
653    updated_at timestamptz NOT NULL DEFAULT now());
654-- assist_marks: a player switching the training app's engine help on or
655-- off during one of our games, as declared by the app. Ground truth for
656-- training the fair-play model; screening never relies on it.
657CREATE TABLE chess.assist_marks (
658    id bigserial PRIMARY KEY,
659    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
660    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
661    ply integer NOT NULL,
662    assist boolean NOT NULL,
663    at timestamptz NOT NULL DEFAULT now());
664CREATE INDEX assist_marks_game_idx ON chess.assist_marks (game_id, player_id);
665ALTER TABLE chess.fairplay_reports
666    ADD COLUMN app_disagree integer,
667    ADD COLUMN app_picked integer,
668    ADD COLUMN app_picked_reference integer,
669    ADD COLUMN declared_assist boolean NOT NULL DEFAULT false;
670",
671    r"
672-- 3. Board photos that couldn't be read for sure, kept for review.
673--
674-- reading: what our board reader made of it (placement, confidence, unsure
675-- squares, model); suggestion: the vision language model's reading, which
676-- the Review page pre-fills; label: the position a person confirmed (FEN
677-- placement, White's side), by the player who sent it in the app or by a
678-- reviewer on the Review page. Labelled photos are training data.
679CREATE TABLE chess.board_photos (
680    id bigserial PRIMARY KEY,
681    player_id bigint,
682    created_at timestamptz NOT NULL DEFAULT now(),
683    image bytea NOT NULL,
684    mime text NOT NULL,
685    reason text NOT NULL CHECK (reason IN ('unsure', 'implausible', 'no_model')),
686    reading jsonb,
687    suggestion jsonb,
688    suggestion_failed text,
689    label text,
690    label_source text CHECK (label_source IN ('player', 'reviewer')),
691    unreadable boolean NOT NULL DEFAULT false,
692    labelled_by bigint,
693    labelled_at timestamptz);
694CREATE INDEX board_photos_open_idx ON chess.board_photos (id DESC) WHERE label IS NULL AND NOT unreadable;
695",
696    r"
697-- 4. Endgame tablebase answers (up to seven pieces), kept once fetched:
698-- the tablebase part of the position dictionary.
699CREATE TABLE chess.tablebase (
700    key text PRIMARY KEY,
701    answer jsonb NOT NULL,
702    fetched_at timestamptz NOT NULL DEFAULT now());
703",
704    r"
705-- 5. The position dictionary grows.
706--
707-- positions: positions people confirmed from board photos join it before
708-- anyone has evaluated them (score_cp NULL, photos counting them); the
709-- fair-play worker evaluates those when it has no game to screen, scoring
710-- every legal move (moves: [{uci, cp, depth}], from the side to move's
711-- view). eval_lease_until: a worker is on it.
712-- position_hints: analyses the training app sends in, one per position and
713-- player (the deepest kept). Hints, not trusted like the worker's.
714ALTER TABLE chess.positions ALTER COLUMN score_cp DROP NOT NULL;
715ALTER TABLE chess.positions ALTER COLUMN depth SET DEFAULT 0;
716ALTER TABLE chess.positions ALTER COLUMN engine DROP NOT NULL;
717ALTER TABLE chess.positions ADD COLUMN photos integer NOT NULL DEFAULT 0;
718ALTER TABLE chess.positions ADD COLUMN moves jsonb;
719ALTER TABLE chess.positions ADD COLUMN eval_lease_until timestamptz;
720CREATE INDEX positions_unscored_idx ON chess.positions (updated_at) WHERE score_cp IS NULL;
721CREATE TABLE chess.position_hints (
722    key text NOT NULL,
723    player_id bigint NOT NULL,
724    best text,
725    score_cp integer NOT NULL,
726    depth integer NOT NULL,
727    engine text NOT NULL,
728    moves jsonb,
729    at timestamptz NOT NULL DEFAULT now(),
730    PRIMARY KEY (key, player_id));
731",
732    r"
733-- 6. Photo review by everyone, and the photo licence.
734--
735-- players.photo_terms: the photo licence the player accepted (its version),
736-- and photo_public whether they also release their photos as CC0. A photo
737-- is kept for review only under accepted terms, which it records.
738-- photo_votes: what each reviewer says a photo shows (anyone signed in; the
739-- player who sent it votes by confirming it). Two agreeing votes label it
740-- ('consensus'); an admin's or manager's word is final ('admin').
741ALTER TABLE chess.players ADD COLUMN IF NOT EXISTS photo_terms text;
742ALTER TABLE chess.players ADD COLUMN IF NOT EXISTS photo_terms_at timestamptz;
743ALTER TABLE chess.players ADD COLUMN IF NOT EXISTS photo_public boolean NOT NULL DEFAULT false;
744ALTER TABLE chess.board_photos ADD COLUMN terms text;
745ALTER TABLE chess.board_photos ADD COLUMN public boolean NOT NULL DEFAULT false;
746ALTER TABLE chess.board_photos DROP CONSTRAINT IF EXISTS board_photos_label_source_check;
747ALTER TABLE chess.board_photos ADD CONSTRAINT board_photos_label_source_check
748    CHECK (label_source IN ('player', 'reviewer', 'consensus', 'admin'));
749CREATE TABLE chess.photo_votes (
750    photo_id bigint NOT NULL REFERENCES chess.board_photos ON DELETE CASCADE,
751    player_id bigint NOT NULL,
752    placement text,
753    unreadable boolean NOT NULL DEFAULT false,
754    voted_at timestamptz NOT NULL DEFAULT now(),
755    PRIMARY KEY (photo_id, player_id));
756",
757    r"
758-- 7. Accounts of our own.
759--
760-- Roles are this server's alone now: a host site's roles no longer come
761-- with its players, so everyone starts again as a player (CHESS_ADMIN_EMAILS
762-- and `open-chess admin grant` make admins). email_verified_at: an email
763-- someone vouched for (the host site, at sign-in), which CHESS_ADMIN_EMAILS
764-- trusts. Passkeys sign people in instead of a password; an admin's or
765-- manager's rights need a passkey check, recorded on the session or token
766-- it was made in (staff_verified_at); a token made through OAuth never
767-- inherits one (an app approved for a scope mustn't get an admin's rights). Partners are other services' API
768-- access: a named token with partner scopes, made by an admin.
769UPDATE chess.players SET role = 'player', roles = '{}' WHERE role <> 'player' OR roles <> '{}';
770ALTER TABLE chess.players ADD COLUMN email_verified_at timestamptz;
771ALTER TABLE chess.sessions ADD COLUMN staff_verified_at timestamptz;
772ALTER TABLE chess.api_tokens ADD COLUMN staff_verified_at timestamptz;
773CREATE TABLE chess.passkeys (
774    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
775    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
776    user_handle bytea NOT NULL,
777    credential_id bytea NOT NULL UNIQUE,
778    public_key bytea NOT NULL,
779    alg integer NOT NULL,
780    sign_count bigint NOT NULL DEFAULT 0,
781    name text NOT NULL,
782    created_at timestamptz NOT NULL DEFAULT now(),
783    last_used_at timestamptz);
784CREATE INDEX passkeys_player_idx ON chess.passkeys (player_id);
785CREATE TABLE chess.webauthn_challenges (
786    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
787    challenge bytea NOT NULL,
788    purpose text NOT NULL CHECK (purpose IN ('register', 'signin', 'signup', 'staff')),
789    player_id bigint,
790    data jsonb,
791    created_at timestamptz NOT NULL DEFAULT now());
792CREATE TABLE chess.partners (
793    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
794    name text NOT NULL,
795    token_hash text NOT NULL UNIQUE,
796    prefix text NOT NULL,
797    scopes text[] NOT NULL,
798    created_at timestamptz NOT NULL DEFAULT now(),
799    created_by bigint,
800    last_used_at timestamptz,
801    revoked_at timestamptz);
802",
803];
804
805#[cfg(test)]
806mod tests {
807    use super::*;
808
809    #[test]
810    fn neon_direct_host() {
811        assert_eq!(
812            direct_host("ep-cool-1234-pooler.eu-central-1.aws.neon.tech").as_deref(),
813            Some("ep-cool-1234.eu-central-1.aws.neon.tech")
814        );
815        assert_eq!(direct_host("localhost"), None);
816        assert_eq!(direct_host("db.internal"), None);
817    }
818
819    #[test]
820    fn listen_url_replaces_the_pooler_host() {
821        let u = listen_url(
822            "postgresql://u:p@ep-cool-1234-pooler.eu-central-1.aws.neon.tech/neondb?sslmode=require",
823        );
824        assert_eq!(u, "postgresql://u:p@ep-cool-1234.eu-central-1.aws.neon.tech/neondb?sslmode=require");
825        let pg = tokio_postgres::Config::from_str(&u).unwrap();
826        assert_eq!(pg.get_hosts().len(), 1, "one host: the direct one");
827        let local = "postgres://dev:dev@127.0.0.1/chess?sslmode=disable";
828        assert_eq!(listen_url(local), local);
829        assert_eq!(listen_url("host=x user=y"), "host=x user=y");
830    }
831
832    #[test]
833    fn notes_round_trip() {
834        for n in [
835            Note::Game("abcd1234".into()),
836            Note::Chat { game: "g".into(), line: serde_json::json!({"type": "chatLine"}) },
837            Note::User { player: 7, event: serde_json::json!({"type": "gameStart"}) },
838            Note::Resync,
839        ] {
840            let back = Db::decode(&Db::encode(&n)).unwrap();
841            assert_eq!(format!("{back:?}"), format!("{n:?}"));
842        }
843        assert!(Db::decode("nope").is_none());
844    }
845
846    #[test]
847    fn prefix_replacement_only_touches_table_names() {
848        let q = "SELECT * FROM chess.games g JOIN site.players p ON p.id = g.white_id -- the website. chess.";
849        assert_eq!(
850            prefix_replace(&prefix_replace(q, "chess.", "\"c\"."), "site.", "\"s\"."),
851            "SELECT * FROM \"c\".games g JOIN \"s\".players p ON p.id = g.white_id -- the website. chess."
852        );
853        assert_eq!(prefix_replace("(chess.a,chess.b)", "chess.", "x."), "(x.a,x.b)");
854    }
855
856    #[test]
857    fn schema_prefixes() {
858        assert_eq!(quote_ident("a\"b"), "\"a\"\"b\"");
859        assert!(MIGRATIONS[0].contains("REFERENCES chess.players"));
860    }
861}
open_chess_server/db.rs

open_chess_server/
db.rs
