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 (they have a chess
205    /// account, or are named in any of our tables, or hold a token with a
206    /// chess scope) with their ids, and their unexpired chess tokens, into
207    /// our own tables, linked to the site by `esports_id`; then point our
208    /// tables' references at our players. A site without those tables (a
209    /// stand-alone server) has nothing to import.
210    async fn import_site(&self, tx: &tokio_postgres::Transaction<'_>) -> Result<()> {
211        if *self.site == *self.schema {
212            return Ok(());
213        }
214        let has = |t: &'static str| {
215            let site = self.site.clone();
216            async move {
217                tx.query_one(
218                    "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2)",
219                    &[&&*site, &t],
220                )
221                .await
222                .map(|r| r.get::<_, bool>(0))
223            }
224        };
225        if !(has("players").await? && has("api_tokens").await?) {
226            return Ok(());
227        }
228        let done = tx
229            .query_opt(&self.sql("SELECT 1 FROM chess.imports WHERE source = $1"), &[&&*self.site])
230            .await?
231            .is_some();
232        if done {
233            return Ok(());
234        }
235        let site_players = format!("{}.players", quote_ident(&self.site));
236        // Our foreign keys still pointing at the site's players.
237        let fks = tx
238            .query(
239                "SELECT c.conname, c.conrelid::regclass::text, a.attname, pg_get_constraintdef(c.oid)
240                   FROM pg_constraint c
241                   JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = c.conkey[1]
242                  WHERE c.contype = 'f' AND c.confrelid = to_regclass($1)
243                    AND c.connamespace = (SELECT oid FROM pg_namespace WHERE nspname = $2)",
244                &[&site_players, &&*self.schema],
245            )
246            .await?;
247        let scopes: Vec<String> = crate::auth::SCOPES.iter().map(|(s, _)| (*s).to_owned()).collect();
248        let mut named = String::from("SELECT player_id FROM site.api_tokens WHERE scopes && $1::text[]");
249        for f in &fks {
250            let (table, column): (String, String) = (f.get(1), f.get(2));
251            let _ = write!(named, " UNION SELECT {} FROM {table}", quote_ident(&column));
252        }
253        let players = tx
254            .execute(
255                &self.sql(&format!(
256                    "INSERT INTO chess.players (id, name, email, key_hash, role, roles, disabled_at, esports_id)
257                     SELECT p.id, p.name, p.email, p.key_hash, p.role, p.roles, p.disabled_at, p.id::text
258                       FROM site.players p WHERE p.id IN ({named})
259                     ON CONFLICT (id) DO NOTHING"
260                )),
261                &[&scopes],
262            )
263            .await?;
264        let tokens = tx
265            .execute(
266                &self.sql(
267                    "INSERT INTO chess.api_tokens (player_id, name, token_hash, prefix, scopes, created_at,
268                                                   expires_at, last_used_at, revoked_at, revoked_by)
269                     SELECT t.player_id, t.name, t.token_hash, t.prefix,
270                            ARRAY(SELECT unnest(t.scopes) INTERSECT SELECT unnest($1::text[])),
271                            t.created_at, t.expires_at, t.last_used_at, t.revoked_at, t.revoked_by
272                       FROM site.api_tokens t
273                      WHERE t.player_id IN (SELECT id FROM chess.players) AND t.scopes && $1::text[]
274                        AND t.revoked_at IS NULL AND t.expires_at > now()
275                     ON CONFLICT (token_hash) DO NOTHING",
276                ),
277                &[&scopes],
278            )
279            .await?;
280        tx.execute(
281            &self.sql(
282                "SELECT setval(pg_get_serial_sequence($1, 'id'), GREATEST((SELECT max(id) FROM chess.players), 1))",
283            ),
284            &[&self.sql("chess.players")],
285        )
286        .await?;
287        let ours = self.sql("chess.players");
288        for f in &fks {
289            let (name, table, column, def): (String, String, String, String) =
290                (f.get(0), f.get(1), f.get(2), f.get(3));
291            let tail = def.split_once(".players(id)").map_or("", |(_, t)| t);
292            tx.batch_execute(&format!(
293                "ALTER TABLE {table} DROP CONSTRAINT {n};
294                 ALTER TABLE {table} ADD CONSTRAINT {n} FOREIGN KEY ({c}) REFERENCES {ours}(id){tail};",
295                n = quote_ident(&name),
296                c = quote_ident(&column),
297            ))
298            .await?;
299        }
300        tx.execute(&self.sql("INSERT INTO chess.imports (source) VALUES ($1)"), &[&&*self.site]).await?;
301        info!(
302            players,
303            tokens,
304            references = fks.len(),
305            site = &*self.site,
306            "imported the site's chess players"
307        );
308        Ok(())
309    }
310
311    /// Keep a LISTEN connection open and forward notifications into the
312    /// local broadcast. Reconnects with backoff; after a reconnect every
313    /// stream is told to resync.
314    pub async fn listen(self, cfg: Config, mut stop: tokio::sync::watch::Receiver<bool>) {
315        let mut backoff = Duration::from_millis(500);
316        let mut first = true;
317        loop {
318            let started = std::time::Instant::now();
319            match self.listen_once(&cfg, first).await {
320                Ok(()) => {}
321                Err(e) => warn!("chess LISTEN connection: {e:#}"),
322            }
323            first = false;
324            if started.elapsed() > Duration::from_secs(30) {
325                backoff = Duration::from_millis(500);
326            }
327            tokio::select! {
328                () = tokio::time::sleep(backoff) => {}
329                () = crate::stopped(&mut stop) => return,
330            }
331            backoff = (backoff * 2).min(Duration::from_secs(10));
332        }
333    }
334
335    async fn listen_once(&self, cfg: &Config, first: bool) -> Result<()> {
336        let url = match &cfg.listen_database_url {
337            Some(u) => u.clone(),
338            None => listen_url(&cfg.database_url),
339        };
340        let pg = tokio_postgres::Config::from_str(&url).context("the LISTEN database URL")?;
341        let (client, mut conn) = pg.connect(tls()?).await?;
342        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
343        let driver = tokio::spawn(async move {
344            let mut s = futures_util::stream::poll_fn(move |cx| conn.poll_message(cx));
345            while let Some(m) = s.next().await {
346                match m {
347                    Ok(tokio_postgres::AsyncMessage::Notification(n)) => {
348                        if tx.send(n.payload().to_owned()).is_err() {
349                            break;
350                        }
351                    }
352                    Ok(_) => {}
353                    Err(e) => {
354                        warn!("chess LISTEN: {e}");
355                        break;
356                    }
357                }
358            }
359        });
360        client.batch_execute(&format!("LISTEN {}", quote_ident(&self.channel))).await?;
361        info!(channel = &*self.channel, "listening for chess changes");
362        if !first {
363            let _ = self.notes.send(Note::Resync);
364        }
365        let mut ping = tokio::time::interval(Duration::from_secs(30));
366        loop {
367            tokio::select! {
368                m = rx.recv() => match m {
369                    Some(p) => {
370                        if let Some(n) = Db::decode(&p) {
371                            let _ = self.notes.send(n);
372                        }
373                    }
374                    None => break,
375                },
376                _ = ping.tick() => {
377                    if client.simple_query("SELECT 1").await.is_err() {
378                        break;
379                    }
380                }
381            }
382        }
383        driver.abort();
384        bail!("the LISTEN connection closed")
385    }
386}
387
388/// The URL for LISTEN: a Neon `-pooler` host swapped for the direct one
389/// (the host is replaced, not added: tokio-postgres would try a second
390/// host only if the first failed, and the pooler doesn't fail, it just
391/// never delivers notifications). Anything else is used as given.
392pub fn listen_url(database_url: &str) -> String {
393    let Ok(mut u) = url::Url::parse(database_url) else { return database_url.to_owned() };
394    let Some(direct) = u.host_str().and_then(direct_host) else { return database_url.to_owned() };
395    if u.set_host(Some(&direct)).is_err() {
396        return database_url.to_owned();
397    }
398    u.into()
399}
400
401/// Replace `from` with `to` wherever it starts a word and is followed by a
402/// letter (a table name).
403fn prefix_replace(q: &str, from: &str, to: &str) -> String {
404    let mut out = String::with_capacity(q.len() + 32);
405    let mut rest = q;
406    let mut prev: Option<char> = None;
407    while let Some(i) = rest.find(from) {
408        let before = rest[..i].chars().next_back().or(prev);
409        let after = rest[i + from.len()..].chars().next();
410        out.push_str(&rest[..i]);
411        let starts_word = before.is_none_or(|c| !(c.is_alphanumeric() || c == '_' || c == '"' || c == '.'));
412        if starts_word && after.is_some_and(|c| c.is_ascii_alphabetic()) {
413            out.push_str(to);
414        } else {
415            out.push_str(from);
416        }
417        prev = from.chars().next_back();
418        rest = &rest[i + from.len()..];
419    }
420    out.push_str(rest);
421    out
422}
423
424/// Send `note` from inside a transaction (delivered to every replica on
425/// commit). Payloads over Postgres's 8000-byte limit are trimmed to a
426/// reload instruction.
427pub async fn notify(tx: &impl tokio_postgres::GenericClient, db: &Db, note: &Note) -> Result<()> {
428    let mut payload = Db::encode(note);
429    if payload.len() > 7900 {
430        payload = match note {
431            Note::Game(id) | Note::Chat { game: id, .. } => Db::encode(&Note::Game(id.clone())),
432            _ => Db::encode(&Note::Resync),
433        };
434    }
435    // Postgres delivers on commit, to every replica listening.
436    tx.execute("SELECT pg_notify($1, $2)", &[&db.channel(), &payload]).await?;
437    Ok(())
438}
439
440/// The server's own players, API tokens and sessions (created before the
441/// migrations, which refer to them; idempotent). `esports_id` links a
442/// player to the host site's (signed in with it, or imported from it).
443const IDENTITY: &str = r"
444CREATE TABLE IF NOT EXISTS chess.players (
445    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
446    name text NOT NULL,
447    email text,
448    key_hash text,
449    password_hash text,
450    role text NOT NULL DEFAULT 'player',
451    roles text[] NOT NULL DEFAULT '{}',
452    esports_id text UNIQUE,
453    created_at timestamptz NOT NULL DEFAULT now(),
454    disabled_at timestamptz);
455CREATE TABLE IF NOT EXISTS chess.api_tokens (
456    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
457    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
458    name text NOT NULL,
459    token_hash text NOT NULL UNIQUE,
460    prefix text NOT NULL,
461    scopes text[] NOT NULL,
462    created_at timestamptz NOT NULL DEFAULT now(),
463    expires_at timestamptz NOT NULL,
464    last_used_at timestamptz,
465    revoked_at timestamptz,
466    revoked_by text);
467CREATE INDEX IF NOT EXISTS api_tokens_player_idx ON chess.api_tokens (player_id);
468CREATE TABLE IF NOT EXISTS chess.sessions (
469    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
470    token_hash text NOT NULL UNIQUE,
471    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
472    created_at timestamptz NOT NULL DEFAULT now(),
473    expires_at timestamptz NOT NULL,
474    revoked_at timestamptz);
475CREATE TABLE IF NOT EXISTS chess.imports (
476    source text PRIMARY KEY,
477    at timestamptz NOT NULL DEFAULT now());
478";
479
480const MIGRATIONS: &[&str] = &[
481    r"
482-- One row per site player who has used the chess server. `username` is
483-- their Lichess-style handle here; `kind` is human or bot (bots play
484-- through the Bot API and are never fair-play screened); `marked_at` is a
485-- confirmed fair-play violation (shown as tosViolation).
486CREATE TABLE chess.accounts (
487    player_id bigint PRIMARY KEY REFERENCES chess.players ON DELETE CASCADE,
488    username text NOT NULL,
489    kind text NOT NULL DEFAULT 'human' CHECK (kind IN ('human', 'bot')),
490    created_at timestamptz NOT NULL DEFAULT now(),
491    seen_at timestamptz,
492    marked_at timestamptz,
493    marked_by bigint,
494    CHECK (username ~ '^[A-Za-z0-9][A-Za-z0-9_-]{1,29}$'));
495CREATE UNIQUE INDEX accounts_username_idx ON chess.accounts (lower(username));
496
497-- Glicko-2 per rating pool (bullet, blitz, …).
498CREATE TABLE chess.ratings (
499    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
500    perf text NOT NULL,
501    rating double precision NOT NULL,
502    deviation double precision NOT NULL,
503    volatility double precision NOT NULL,
504    games integer NOT NULL DEFAULT 0,
505    prog integer NOT NULL DEFAULT 0,
506    updated_at timestamptz NOT NULL DEFAULT now(),
507    PRIMARY KEY (player_id, perf));
508
509CREATE TABLE chess.challenges (
510    id text PRIMARY KEY,
511    challenger_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
512    dest_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
513    status text NOT NULL DEFAULT 'created'
514        CHECK (status IN ('created', 'offline', 'canceled', 'declined', 'accepted')),
515    variant text NOT NULL DEFAULT 'standard' CHECK (variant IN ('standard', 'fromPosition')),
516    initial_fen text,
517    rated boolean NOT NULL DEFAULT false,
518    clock_limit integer,
519    clock_increment integer,
520    color text NOT NULL DEFAULT 'random' CHECK (color IN ('white', 'black', 'random')),
521    final_color text NOT NULL CHECK (final_color IN ('white', 'black')),
522    rules text[] NOT NULL DEFAULT '{}',
523    decline_reason text,
524    created_at timestamptz NOT NULL DEFAULT now(),
525    expires_at timestamptz NOT NULL);
526CREATE INDEX challenges_dest_idx ON chess.challenges (dest_id) WHERE status = 'created';
527CREATE INDEX challenges_challenger_idx ON chess.challenges (challenger_id) WHERE status = 'created';
528CREATE INDEX challenges_expiry_idx ON chess.challenges (expires_at) WHERE status = 'created';
529
530-- Games. `moves` is UCI (standard castling), `turn_started_at` is when the
531-- side to move's clock started (the database clock is the only clock, so
532-- every replica agrees). move_ms / clock_ms: time used and time left per
533-- ply, for exports and fair-play timing.
534CREATE TABLE chess.games (
535    id text PRIMARY KEY,
536    white_id bigint NOT NULL REFERENCES chess.accounts,
537    black_id bigint NOT NULL REFERENCES chess.accounts,
538    variant text NOT NULL DEFAULT 'standard',
539    initial_fen text,
540    rated boolean NOT NULL,
541    source text NOT NULL DEFAULT 'friend',
542    speed text NOT NULL,
543    rules text[] NOT NULL DEFAULT '{}',
544    clock_initial_ms bigint,
545    clock_increment_ms bigint,
546    status smallint NOT NULL DEFAULT 20,
547    winner text CHECK (winner IN ('white', 'black')),
548    moves text NOT NULL DEFAULT '',
549    ply integer NOT NULL DEFAULT 0,
550    fen text NOT NULL,
551    white_ms bigint,
552    black_ms bigint,
553    move_ms bigint[] NOT NULL DEFAULT '{}',
554    clock_ms bigint[] NOT NULL DEFAULT '{}',
555    draw_offer text CHECK (draw_offer IN ('white', 'black')),
556    takeback_offer text CHECK (takeback_offer IN ('white', 'black')),
557    white_rating integer NOT NULL,
558    black_rating integer NOT NULL,
559    white_provisional boolean NOT NULL DEFAULT false,
560    black_provisional boolean NOT NULL DEFAULT false,
561    white_rating_diff integer,
562    black_rating_diff integer,
563    created_at timestamptz NOT NULL DEFAULT now(),
564    turn_started_at timestamptz NOT NULL DEFAULT now(),
565    last_move_at timestamptz,
566    finished_at timestamptz,
567    analysis text NOT NULL DEFAULT 'none'
568        CHECK (analysis IN ('none', 'pending', 'running', 'done', 'failed', 'skipped')),
569    analysis_lease_until timestamptz,
570    CHECK (white_id <> black_id));
571CREATE INDEX games_ongoing_idx ON chess.games (turn_started_at) WHERE status IN (10, 20);
572CREATE INDEX games_white_idx ON chess.games (white_id, created_at DESC);
573CREATE INDEX games_black_idx ON chess.games (black_id, created_at DESC);
574CREATE INDEX games_analysis_idx ON chess.games (finished_at) WHERE analysis IN ('pending', 'running');
575
576CREATE TABLE chess.chat (
577    id bigserial PRIMARY KEY,
578    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
579    room text NOT NULL CHECK (room IN ('player', 'spectator')),
580    player_id bigint,
581    username text NOT NULL,
582    text text NOT NULL CHECK (length(text) <= 140),
583    at timestamptz NOT NULL DEFAULT now());
584CREATE INDEX chat_game_idx ON chess.chat (game_id, id);
585
586-- Who has a game stream open (heartbeats; nothing here needs to survive
587-- a restart, hence unlogged).
588CREATE UNLOGGED TABLE chess.presence (
589    game_id text NOT NULL,
590    player_id bigint NOT NULL,
591    seen_at timestamptz NOT NULL DEFAULT now(),
592    PRIMARY KEY (game_id, player_id));
593
594-- OAuth authorization codes (PKCE), single use, two minutes.
595CREATE TABLE chess.oauth_codes (
596    code_hash text PRIMARY KEY,
597    player_id bigint NOT NULL REFERENCES chess.players ON DELETE CASCADE,
598    client_id text NOT NULL,
599    redirect_uri text NOT NULL,
600    code_challenge text NOT NULL,
601    scopes text[] NOT NULL,
602    expires_at timestamptz NOT NULL,
603    used_at timestamptz);
604
605-- Fair play: one report per screened side of a game, and reviewers'
606-- verdicts.
607CREATE TABLE chess.fairplay_reports (
608    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
609    side text NOT NULL CHECK (side IN ('white', 'black')),
610    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
611    score real NOT NULL,
612    flagged boolean NOT NULL,
613    moves integer NOT NULL,
614    accuracy real NOT NULL,
615    acpl real NOT NULL,
616    top1 real NOT NULL,
617    top3 real NOT NULL,
618    time_cv real,
619    time_corr real,
620    reasons text[] NOT NULL DEFAULT '{}',
621    engine text NOT NULL,
622    depth integer NOT NULL,
623    evals jsonb NOT NULL,
624    created_at timestamptz NOT NULL DEFAULT now(),
625    PRIMARY KEY (game_id, side));
626CREATE INDEX fairplay_flagged_idx ON chess.fairplay_reports (created_at DESC) WHERE flagged;
627CREATE INDEX fairplay_player_idx ON chess.fairplay_reports (player_id, created_at DESC);
628CREATE TABLE chess.fairplay_reviews (
629    id bigserial PRIMARY KEY,
630    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
631    game_id text REFERENCES chess.games ON DELETE SET NULL,
632    verdict text NOT NULL CHECK (verdict IN ('clean', 'engine', 'unclear')),
633    note text,
634    reviewer_id bigint NOT NULL,
635    at timestamptz NOT NULL DEFAULT now());
636",
637    r"
638-- 2. The position dictionary, assistance declarations, and the training-app
639-- fingerprint on fair-play reports.
640--
641-- positions: one row per position (FEN without the move counters), the
642-- deepest engine evaluation stored for it, filled by the fair-play worker
643-- as it screens games. A lookup is a key read.
644CREATE TABLE chess.positions (
645    key text PRIMARY KEY,
646    best text,
647    score_cp integer NOT NULL,
648    depth integer NOT NULL,
649    engine text NOT NULL,
650    seen integer NOT NULL DEFAULT 1,
651    updated_at timestamptz NOT NULL DEFAULT now());
652-- assist_marks: a player switching the training app's engine help on or
653-- off during one of our games, as declared by the app. Ground truth for
654-- training the fair-play model; screening never relies on it.
655CREATE TABLE chess.assist_marks (
656    id bigserial PRIMARY KEY,
657    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
658    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
659    ply integer NOT NULL,
660    assist boolean NOT NULL,
661    at timestamptz NOT NULL DEFAULT now());
662CREATE INDEX assist_marks_game_idx ON chess.assist_marks (game_id, player_id);
663ALTER TABLE chess.fairplay_reports
664    ADD COLUMN app_disagree integer,
665    ADD COLUMN app_picked integer,
666    ADD COLUMN app_picked_reference integer,
667    ADD COLUMN declared_assist boolean NOT NULL DEFAULT false;
668",
669    r"
670-- 3. Board photos that couldn't be read for sure, kept for review.
671--
672-- reading: what our board reader made of it (placement, confidence, unsure
673-- squares, model); suggestion: the vision language model's reading, which
674-- the Review page pre-fills; label: the position a person confirmed (FEN
675-- placement, White's side), by the player who sent it in the app or by a
676-- reviewer on the Review page. Labelled photos are training data.
677CREATE TABLE chess.board_photos (
678    id bigserial PRIMARY KEY,
679    player_id bigint,
680    created_at timestamptz NOT NULL DEFAULT now(),
681    image bytea NOT NULL,
682    mime text NOT NULL,
683    reason text NOT NULL CHECK (reason IN ('unsure', 'implausible', 'no_model')),
684    reading jsonb,
685    suggestion jsonb,
686    suggestion_failed text,
687    label text,
688    label_source text CHECK (label_source IN ('player', 'reviewer')),
689    unreadable boolean NOT NULL DEFAULT false,
690    labelled_by bigint,
691    labelled_at timestamptz);
692CREATE INDEX board_photos_open_idx ON chess.board_photos (id DESC) WHERE label IS NULL AND NOT unreadable;
693",
694    r"
695-- 4. Endgame tablebase answers (up to seven pieces), kept once fetched:
696-- the tablebase part of the position dictionary.
697CREATE TABLE chess.tablebase (
698    key text PRIMARY KEY,
699    answer jsonb NOT NULL,
700    fetched_at timestamptz NOT NULL DEFAULT now());
701",
702    r"
703-- 5. The position dictionary grows.
704--
705-- positions: positions people confirmed from board photos join it before
706-- anyone has evaluated them (score_cp NULL, photos counting them); the
707-- fair-play worker evaluates those when it has no game to screen, scoring
708-- every legal move (moves: [{uci, cp, depth}], from the side to move's
709-- view). eval_lease_until: a worker is on it.
710-- position_hints: analyses the training app sends in, one per position and
711-- player (the deepest kept). Hints, not trusted like the worker's.
712ALTER TABLE chess.positions ALTER COLUMN score_cp DROP NOT NULL;
713ALTER TABLE chess.positions ALTER COLUMN depth SET DEFAULT 0;
714ALTER TABLE chess.positions ALTER COLUMN engine DROP NOT NULL;
715ALTER TABLE chess.positions ADD COLUMN photos integer NOT NULL DEFAULT 0;
716ALTER TABLE chess.positions ADD COLUMN moves jsonb;
717ALTER TABLE chess.positions ADD COLUMN eval_lease_until timestamptz;
718CREATE INDEX positions_unscored_idx ON chess.positions (updated_at) WHERE score_cp IS NULL;
719CREATE TABLE chess.position_hints (
720    key text NOT NULL,
721    player_id bigint NOT NULL,
722    best text,
723    score_cp integer NOT NULL,
724    depth integer NOT NULL,
725    engine text NOT NULL,
726    moves jsonb,
727    at timestamptz NOT NULL DEFAULT now(),
728    PRIMARY KEY (key, player_id));
729",
730    r"
731-- 6. Photo review by everyone, and the photo licence.
732--
733-- players.photo_terms: the photo licence the player accepted (its version),
734-- and photo_public whether they also release their photos as CC0. A photo
735-- is kept for review only under accepted terms, which it records.
736-- photo_votes: what each reviewer says a photo shows (anyone signed in; the
737-- player who sent it votes by confirming it). Two agreeing votes label it
738-- ('consensus'); an admin's or manager's word is final ('admin').
739ALTER TABLE chess.players ADD COLUMN IF NOT EXISTS photo_terms text;
740ALTER TABLE chess.players ADD COLUMN IF NOT EXISTS photo_terms_at timestamptz;
741ALTER TABLE chess.players ADD COLUMN IF NOT EXISTS photo_public boolean NOT NULL DEFAULT false;
742ALTER TABLE chess.board_photos ADD COLUMN terms text;
743ALTER TABLE chess.board_photos ADD COLUMN public boolean NOT NULL DEFAULT false;
744ALTER TABLE chess.board_photos DROP CONSTRAINT IF EXISTS board_photos_label_source_check;
745ALTER TABLE chess.board_photos ADD CONSTRAINT board_photos_label_source_check
746    CHECK (label_source IN ('player', 'reviewer', 'consensus', 'admin'));
747CREATE TABLE chess.photo_votes (
748    photo_id bigint NOT NULL REFERENCES chess.board_photos ON DELETE CASCADE,
749    player_id bigint NOT NULL,
750    placement text,
751    unreadable boolean NOT NULL DEFAULT false,
752    voted_at timestamptz NOT NULL DEFAULT now(),
753    PRIMARY KEY (photo_id, player_id));
754",
755];
756
757#[cfg(test)]
758mod tests {
759    use super::*;
760
761    #[test]
762    fn neon_direct_host() {
763        assert_eq!(
764            direct_host("ep-cool-1234-pooler.eu-central-1.aws.neon.tech").as_deref(),
765            Some("ep-cool-1234.eu-central-1.aws.neon.tech")
766        );
767        assert_eq!(direct_host("localhost"), None);
768        assert_eq!(direct_host("db.internal"), None);
769    }
770
771    #[test]
772    fn listen_url_replaces_the_pooler_host() {
773        let u = listen_url(
774            "postgresql://u:p@ep-cool-1234-pooler.eu-central-1.aws.neon.tech/neondb?sslmode=require",
775        );
776        assert_eq!(u, "postgresql://u:p@ep-cool-1234.eu-central-1.aws.neon.tech/neondb?sslmode=require");
777        let pg = tokio_postgres::Config::from_str(&u).unwrap();
778        assert_eq!(pg.get_hosts().len(), 1, "one host: the direct one");
779        let local = "postgres://dev:dev@127.0.0.1/chess?sslmode=disable";
780        assert_eq!(listen_url(local), local);
781        assert_eq!(listen_url("host=x user=y"), "host=x user=y");
782    }
783
784    #[test]
785    fn notes_round_trip() {
786        for n in [
787            Note::Game("abcd1234".into()),
788            Note::Chat { game: "g".into(), line: serde_json::json!({"type": "chatLine"}) },
789            Note::User { player: 7, event: serde_json::json!({"type": "gameStart"}) },
790            Note::Resync,
791        ] {
792            let back = Db::decode(&Db::encode(&n)).unwrap();
793            assert_eq!(format!("{back:?}"), format!("{n:?}"));
794        }
795        assert!(Db::decode("nope").is_none());
796    }
797
798    #[test]
799    fn prefix_replacement_only_touches_table_names() {
800        let q = "SELECT * FROM chess.games g JOIN site.players p ON p.id = g.white_id -- the website. chess.";
801        assert_eq!(
802            prefix_replace(&prefix_replace(q, "chess.", "\"c\"."), "site.", "\"s\"."),
803            "SELECT * FROM \"c\".games g JOIN \"s\".players p ON p.id = g.white_id -- the website. chess."
804        );
805        assert_eq!(prefix_replace("(chess.a,chess.b)", "chess.", "x."), "(x.a,x.b)");
806    }
807
808    #[test]
809    fn schema_prefixes() {
810        assert_eq!(quote_ident("a\"b"), "\"a\"\"b\"");
811        assert!(MIGRATIONS[0].contains("REFERENCES chess.players"));
812    }
813}
open_chess_server/db.rs

open_chess_server/
db.rs
