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.` and `site.` table prefixes; [`Db::sql`]
5//! swaps in the configured schema names. Explicit names rather than a
6//! `search_path` because Neon's pooler (transaction pooling) doesn't keep
7//! session settings.
8
9use std::str::FromStr;
10use std::sync::Arc;
11use std::time::Duration;
12
13use anyhow::{Context, Result, bail};
14use deadpool::managed::{self, Metrics, RecycleError, RecycleResult};
15use futures_util::StreamExt;
16use tokio::sync::broadcast;
17use tracing::{info, warn};
18
19use crate::config::Config;
20
21pub fn quote_ident(s: &str) -> String {
22    format!("\"{}\"", s.replace('"', "\"\""))
23}
24
25fn tls() -> Result<tokio_postgres_rustls::MakeRustlsConnect> {
26    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
27    let mut roots = rustls::RootCertStore::empty();
28    let native = rustls_native_certs::load_native_certs();
29    let (added, _) = roots.add_parsable_certificates(native.certs);
30    if added == 0 {
31        bail!("no trusted root certificates found for TLS");
32    }
33    let cfg = rustls::ClientConfig::builder().with_root_certificates(roots).with_no_client_auth();
34    Ok(tokio_postgres_rustls::MakeRustlsConnect::new(cfg))
35}
36
37/// Neon's pooled hostname (`ep-x-123-pooler.region…`) → the direct one,
38/// which LISTEN needs (transaction poolers don't deliver notifications).
39pub fn direct_host(host: &str) -> Option<String> {
40    let (first, rest) = host.split_once('.')?;
41    first.strip_suffix("-pooler").map(|ep| format!("{ep}.{rest}"))
42}
43
44pub struct Manager {
45    pg: tokio_postgres::Config,
46    tls: tokio_postgres_rustls::MakeRustlsConnect,
47}
48
49impl managed::Manager for Manager {
50    type Type = tokio_postgres::Client;
51    type Error = tokio_postgres::Error;
52
53    async fn create(&self) -> Result<tokio_postgres::Client, tokio_postgres::Error> {
54        let (client, conn) = self.pg.connect(self.tls.clone()).await?;
55        tokio::spawn(async move {
56            if let Err(e) = conn.await {
57                warn!("postgres connection: {e}");
58            }
59        });
60        Ok(client)
61    }
62
63    async fn recycle(
64        &self,
65        client: &mut tokio_postgres::Client,
66        _: &Metrics,
67    ) -> RecycleResult<tokio_postgres::Error> {
68        if client.is_closed() {
69            return Err(RecycleError::message("connection closed"));
70        }
71        Ok(())
72    }
73}
74
75pub type Pool = managed::Pool<Manager>;
76pub type Conn = managed::Object<Manager>;
77
78/// A change other replicas (and local streams) need to hear about.
79#[derive(Debug, Clone)]
80pub enum Note {
81    /// A game's state changed: reload it.
82    Game(String),
83    /// A chat line in a game.
84    Chat { game: String, line: serde_json::Value },
85    /// An event for one user's event stream (already in its final JSON).
86    User { player: i64, event: serde_json::Value },
87    /// The LISTEN connection dropped and came back: notes may have been
88    /// missed, so every stream reloads.
89    Resync,
90}
91
92#[derive(Clone)]
93pub struct Db {
94    pub pool: Pool,
95    schema: Arc<str>,
96    site: Arc<str>,
97    channel: Arc<str>,
98    notes: broadcast::Sender<Note>,
99}
100
101impl Db {
102    pub fn connect(cfg: &Config) -> Result<Db> {
103        let pg = tokio_postgres::Config::from_str(&cfg.database_url)
104            .context("DATABASE_URL isn't a postgres:// URL")?;
105        let pool = Pool::builder(Manager { pg, tls: tls()? })
106            .max_size(cfg.pool_size.max(2))
107            .wait_timeout(Some(Duration::from_secs(10)))
108            .create_timeout(Some(Duration::from_secs(15)))
109            .runtime(deadpool::Runtime::Tokio1)
110            .build()
111            .context("building the database pool")?;
112        let (notes, _) = broadcast::channel(4096);
113        Ok(Db {
114            pool,
115            schema: cfg.schema.as_str().into(),
116            site: cfg.site_schema.as_str().into(),
117            channel: format!("{}_events", cfg.schema).into(),
118            notes,
119        })
120    }
121
122    /// Swap `chess.` and `site.` table prefixes for the configured schemas
123    /// (only where a prefix starts a word, so prose in comments is safe).
124    pub fn sql(&self, q: &str) -> String {
125        let c = format!("{}.", quote_ident(&self.schema));
126        let s = format!("{}.", quote_ident(&self.site));
127        prefix_replace(&prefix_replace(q, "chess.", &c), "site.", &s)
128    }
129
130    pub async fn conn(&self) -> Result<Conn> {
131        self.pool.get().await.map_err(|e| anyhow::anyhow!("database unavailable: {e}"))
132    }
133
134    pub fn channel(&self) -> &str {
135        &self.channel
136    }
137
138    pub fn subscribe(&self) -> broadcast::Receiver<Note> {
139        self.notes.subscribe()
140    }
141
142    pub fn encode(n: &Note) -> String {
143        let v = match n {
144            Note::Game(id) => serde_json::json!({"t": "g", "g": id}),
145            Note::Chat { game, line } => serde_json::json!({"t": "c", "g": game, "l": line}),
146            Note::User { player, event } => serde_json::json!({"t": "u", "u": player, "e": event}),
147            Note::Resync => serde_json::json!({"t": "r"}),
148        };
149        v.to_string()
150    }
151
152    pub fn decode(s: &str) -> Option<Note> {
153        let v: serde_json::Value = serde_json::from_str(s).ok()?;
154        match v.get("t")?.as_str()? {
155            "g" => Some(Note::Game(v.get("g")?.as_str()?.to_owned())),
156            "c" => Some(Note::Chat { game: v.get("g")?.as_str()?.to_owned(), line: v.get("l")?.clone() }),
157            "u" => Some(Note::User { player: v.get("u")?.as_i64()?, event: v.get("e")?.clone() }),
158            "r" => Some(Note::Resync),
159            _ => None,
160        }
161    }
162
163    /// Run migrations: this server's schema and tables. The site's schema
164    /// must exist (its players and API tokens are what we sign people in
165    /// with).
166    pub async fn migrate(&self) -> Result<()> {
167        let mut c = self.conn().await?;
168        let site_ok: bool = c
169            .query_one(
170                "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = 'players')
171                    AND EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = 'api_tokens')",
172                &[&&*self.site],
173            )
174            .await?
175            .get(0);
176        if !site_ok {
177            bail!(
178                "schema {:?} has no players / api_tokens tables: point CHESS_SITE_SCHEMA at the site's schema \
179                 (run the site once to create it)",
180                &*self.site
181            );
182        }
183        let tx = c.transaction().await?;
184        tx.batch_execute(&format!(
185            "SELECT pg_advisory_xact_lock(hashtext('open-chess-migrate'));
186             CREATE SCHEMA IF NOT EXISTS {s};
187             CREATE TABLE IF NOT EXISTS {s}.schema_version (version integer NOT NULL);",
188            s = quote_ident(&self.schema)
189        ))
190        .await?;
191        let current: i32 = tx
192            .query_opt(&self.sql("SELECT version FROM chess.schema_version"), &[])
193            .await?
194            .map_or(0, |r| r.get(0));
195        for (i, m) in MIGRATIONS.iter().enumerate() {
196            let v = i32::try_from(i + 1).unwrap_or(i32::MAX);
197            if v > current {
198                tx.batch_execute(&self.sql(m))
199                    .await
200                    .with_context(|| format!("chess schema migration {v}"))?;
201                info!(version = v, "chess schema migrated");
202            }
203        }
204        let latest = i32::try_from(MIGRATIONS.len()).unwrap_or(i32::MAX);
205        if current == 0 {
206            tx.execute(&self.sql("INSERT INTO chess.schema_version VALUES ($1)"), &[&latest]).await?;
207        } else if latest > current {
208            tx.execute(&self.sql("UPDATE chess.schema_version SET version = $1"), &[&latest]).await?;
209        }
210        tx.commit().await?;
211        Ok(())
212    }
213
214    /// Keep a LISTEN connection open and forward notifications into the
215    /// local broadcast. Reconnects with backoff; after a reconnect every
216    /// stream is told to resync.
217    pub async fn listen(self, cfg: Config, mut stop: tokio::sync::watch::Receiver<bool>) {
218        let mut backoff = Duration::from_millis(500);
219        let mut first = true;
220        loop {
221            let started = std::time::Instant::now();
222            match self.listen_once(&cfg, first).await {
223                Ok(()) => {}
224                Err(e) => warn!("chess LISTEN connection: {e:#}"),
225            }
226            first = false;
227            if started.elapsed() > Duration::from_secs(30) {
228                backoff = Duration::from_millis(500);
229            }
230            tokio::select! {
231                () = tokio::time::sleep(backoff) => {}
232                () = crate::stopped(&mut stop) => return,
233            }
234            backoff = (backoff * 2).min(Duration::from_secs(10));
235        }
236    }
237
238    async fn listen_once(&self, cfg: &Config, first: bool) -> Result<()> {
239        let url = match &cfg.listen_database_url {
240            Some(u) => u.clone(),
241            None => listen_url(&cfg.database_url),
242        };
243        let pg = tokio_postgres::Config::from_str(&url).context("the LISTEN database URL")?;
244        let (client, mut conn) = pg.connect(tls()?).await?;
245        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
246        let driver = tokio::spawn(async move {
247            let mut s = futures_util::stream::poll_fn(move |cx| conn.poll_message(cx));
248            while let Some(m) = s.next().await {
249                match m {
250                    Ok(tokio_postgres::AsyncMessage::Notification(n)) => {
251                        if tx.send(n.payload().to_owned()).is_err() {
252                            break;
253                        }
254                    }
255                    Ok(_) => {}
256                    Err(e) => {
257                        warn!("chess LISTEN: {e}");
258                        break;
259                    }
260                }
261            }
262        });
263        client.batch_execute(&format!("LISTEN {}", quote_ident(&self.channel))).await?;
264        info!(channel = &*self.channel, "listening for chess changes");
265        if !first {
266            let _ = self.notes.send(Note::Resync);
267        }
268        let mut ping = tokio::time::interval(Duration::from_secs(30));
269        loop {
270            tokio::select! {
271                m = rx.recv() => match m {
272                    Some(p) => {
273                        if let Some(n) = Db::decode(&p) {
274                            let _ = self.notes.send(n);
275                        }
276                    }
277                    None => break,
278                },
279                _ = ping.tick() => {
280                    if client.simple_query("SELECT 1").await.is_err() {
281                        break;
282                    }
283                }
284            }
285        }
286        driver.abort();
287        bail!("the LISTEN connection closed")
288    }
289}
290
291/// The URL for LISTEN: a Neon `-pooler` host swapped for the direct one
292/// (the host is replaced, not added: tokio-postgres would try a second
293/// host only if the first failed, and the pooler doesn't fail, it just
294/// never delivers notifications). Anything else is used as given.
295pub fn listen_url(database_url: &str) -> String {
296    let Ok(mut u) = url::Url::parse(database_url) else { return database_url.to_owned() };
297    let Some(direct) = u.host_str().and_then(direct_host) else { return database_url.to_owned() };
298    if u.set_host(Some(&direct)).is_err() {
299        return database_url.to_owned();
300    }
301    u.into()
302}
303
304/// Replace `from` with `to` wherever it starts a word and is followed by a
305/// letter (a table name).
306fn prefix_replace(q: &str, from: &str, to: &str) -> String {
307    let mut out = String::with_capacity(q.len() + 32);
308    let mut rest = q;
309    let mut prev: Option<char> = None;
310    while let Some(i) = rest.find(from) {
311        let before = rest[..i].chars().next_back().or(prev);
312        let after = rest[i + from.len()..].chars().next();
313        out.push_str(&rest[..i]);
314        let starts_word = before.is_none_or(|c| !(c.is_alphanumeric() || c == '_' || c == '"' || c == '.'));
315        if starts_word && after.is_some_and(|c| c.is_ascii_alphabetic()) {
316            out.push_str(to);
317        } else {
318            out.push_str(from);
319        }
320        prev = from.chars().next_back();
321        rest = &rest[i + from.len()..];
322    }
323    out.push_str(rest);
324    out
325}
326
327/// Send `note` from inside a transaction (delivered to every replica on
328/// commit). Payloads over Postgres's 8000-byte limit are trimmed to a
329/// reload instruction.
330pub async fn notify(tx: &impl tokio_postgres::GenericClient, db: &Db, note: &Note) -> Result<()> {
331    let mut payload = Db::encode(note);
332    if payload.len() > 7900 {
333        payload = match note {
334            Note::Game(id) | Note::Chat { game: id, .. } => Db::encode(&Note::Game(id.clone())),
335            _ => Db::encode(&Note::Resync),
336        };
337    }
338    // Postgres delivers on commit, to every replica listening.
339    tx.execute("SELECT pg_notify($1, $2)", &[&db.channel(), &payload]).await?;
340    Ok(())
341}
342
343const MIGRATIONS: &[&str] = &[
344    r"
345-- One row per site player who has used the chess server. `username` is
346-- their Lichess-style handle here; `kind` is human or bot (bots play
347-- through the Bot API and are never fair-play screened); `marked_at` is a
348-- confirmed fair-play violation (shown as tosViolation).
349CREATE TABLE chess.accounts (
350    player_id bigint PRIMARY KEY REFERENCES site.players ON DELETE CASCADE,
351    username text NOT NULL,
352    kind text NOT NULL DEFAULT 'human' CHECK (kind IN ('human', 'bot')),
353    created_at timestamptz NOT NULL DEFAULT now(),
354    seen_at timestamptz,
355    marked_at timestamptz,
356    marked_by bigint,
357    CHECK (username ~ '^[A-Za-z0-9][A-Za-z0-9_-]{1,29}$'));
358CREATE UNIQUE INDEX accounts_username_idx ON chess.accounts (lower(username));
359
360-- Glicko-2 per rating pool (bullet, blitz, …).
361CREATE TABLE chess.ratings (
362    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
363    perf text NOT NULL,
364    rating double precision NOT NULL,
365    deviation double precision NOT NULL,
366    volatility double precision NOT NULL,
367    games integer NOT NULL DEFAULT 0,
368    prog integer NOT NULL DEFAULT 0,
369    updated_at timestamptz NOT NULL DEFAULT now(),
370    PRIMARY KEY (player_id, perf));
371
372CREATE TABLE chess.challenges (
373    id text PRIMARY KEY,
374    challenger_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
375    dest_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
376    status text NOT NULL DEFAULT 'created'
377        CHECK (status IN ('created', 'offline', 'canceled', 'declined', 'accepted')),
378    variant text NOT NULL DEFAULT 'standard' CHECK (variant IN ('standard', 'fromPosition')),
379    initial_fen text,
380    rated boolean NOT NULL DEFAULT false,
381    clock_limit integer,
382    clock_increment integer,
383    color text NOT NULL DEFAULT 'random' CHECK (color IN ('white', 'black', 'random')),
384    final_color text NOT NULL CHECK (final_color IN ('white', 'black')),
385    rules text[] NOT NULL DEFAULT '{}',
386    decline_reason text,
387    created_at timestamptz NOT NULL DEFAULT now(),
388    expires_at timestamptz NOT NULL);
389CREATE INDEX challenges_dest_idx ON chess.challenges (dest_id) WHERE status = 'created';
390CREATE INDEX challenges_challenger_idx ON chess.challenges (challenger_id) WHERE status = 'created';
391CREATE INDEX challenges_expiry_idx ON chess.challenges (expires_at) WHERE status = 'created';
392
393-- Games. `moves` is UCI (standard castling), `turn_started_at` is when the
394-- side to move's clock started (the database clock is the only clock, so
395-- every replica agrees). move_ms / clock_ms: time used and time left per
396-- ply, for exports and fair-play timing.
397CREATE TABLE chess.games (
398    id text PRIMARY KEY,
399    white_id bigint NOT NULL REFERENCES chess.accounts,
400    black_id bigint NOT NULL REFERENCES chess.accounts,
401    variant text NOT NULL DEFAULT 'standard',
402    initial_fen text,
403    rated boolean NOT NULL,
404    source text NOT NULL DEFAULT 'friend',
405    speed text NOT NULL,
406    rules text[] NOT NULL DEFAULT '{}',
407    clock_initial_ms bigint,
408    clock_increment_ms bigint,
409    status smallint NOT NULL DEFAULT 20,
410    winner text CHECK (winner IN ('white', 'black')),
411    moves text NOT NULL DEFAULT '',
412    ply integer NOT NULL DEFAULT 0,
413    fen text NOT NULL,
414    white_ms bigint,
415    black_ms bigint,
416    move_ms bigint[] NOT NULL DEFAULT '{}',
417    clock_ms bigint[] NOT NULL DEFAULT '{}',
418    draw_offer text CHECK (draw_offer IN ('white', 'black')),
419    takeback_offer text CHECK (takeback_offer IN ('white', 'black')),
420    white_rating integer NOT NULL,
421    black_rating integer NOT NULL,
422    white_provisional boolean NOT NULL DEFAULT false,
423    black_provisional boolean NOT NULL DEFAULT false,
424    white_rating_diff integer,
425    black_rating_diff integer,
426    created_at timestamptz NOT NULL DEFAULT now(),
427    turn_started_at timestamptz NOT NULL DEFAULT now(),
428    last_move_at timestamptz,
429    finished_at timestamptz,
430    analysis text NOT NULL DEFAULT 'none'
431        CHECK (analysis IN ('none', 'pending', 'running', 'done', 'failed', 'skipped')),
432    analysis_lease_until timestamptz,
433    CHECK (white_id <> black_id));
434CREATE INDEX games_ongoing_idx ON chess.games (turn_started_at) WHERE status IN (10, 20);
435CREATE INDEX games_white_idx ON chess.games (white_id, created_at DESC);
436CREATE INDEX games_black_idx ON chess.games (black_id, created_at DESC);
437CREATE INDEX games_analysis_idx ON chess.games (finished_at) WHERE analysis IN ('pending', 'running');
438
439CREATE TABLE chess.chat (
440    id bigserial PRIMARY KEY,
441    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
442    room text NOT NULL CHECK (room IN ('player', 'spectator')),
443    player_id bigint,
444    username text NOT NULL,
445    text text NOT NULL CHECK (length(text) <= 140),
446    at timestamptz NOT NULL DEFAULT now());
447CREATE INDEX chat_game_idx ON chess.chat (game_id, id);
448
449-- Who has a game stream open (heartbeats; nothing here needs to survive
450-- a restart, hence unlogged).
451CREATE UNLOGGED TABLE chess.presence (
452    game_id text NOT NULL,
453    player_id bigint NOT NULL,
454    seen_at timestamptz NOT NULL DEFAULT now(),
455    PRIMARY KEY (game_id, player_id));
456
457-- OAuth authorization codes (PKCE), single use, two minutes.
458CREATE TABLE chess.oauth_codes (
459    code_hash text PRIMARY KEY,
460    player_id bigint NOT NULL REFERENCES site.players ON DELETE CASCADE,
461    client_id text NOT NULL,
462    redirect_uri text NOT NULL,
463    code_challenge text NOT NULL,
464    scopes text[] NOT NULL,
465    expires_at timestamptz NOT NULL,
466    used_at timestamptz);
467
468-- Fair play: one report per screened side of a game, and reviewers'
469-- verdicts.
470CREATE TABLE chess.fairplay_reports (
471    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
472    side text NOT NULL CHECK (side IN ('white', 'black')),
473    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
474    score real NOT NULL,
475    flagged boolean NOT NULL,
476    moves integer NOT NULL,
477    accuracy real NOT NULL,
478    acpl real NOT NULL,
479    top1 real NOT NULL,
480    top3 real NOT NULL,
481    time_cv real,
482    time_corr real,
483    reasons text[] NOT NULL DEFAULT '{}',
484    engine text NOT NULL,
485    depth integer NOT NULL,
486    evals jsonb NOT NULL,
487    created_at timestamptz NOT NULL DEFAULT now(),
488    PRIMARY KEY (game_id, side));
489CREATE INDEX fairplay_flagged_idx ON chess.fairplay_reports (created_at DESC) WHERE flagged;
490CREATE INDEX fairplay_player_idx ON chess.fairplay_reports (player_id, created_at DESC);
491CREATE TABLE chess.fairplay_reviews (
492    id bigserial PRIMARY KEY,
493    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
494    game_id text REFERENCES chess.games ON DELETE SET NULL,
495    verdict text NOT NULL CHECK (verdict IN ('clean', 'engine', 'unclear')),
496    note text,
497    reviewer_id bigint NOT NULL,
498    at timestamptz NOT NULL DEFAULT now());
499",
500    r"
501-- 2. The position dictionary, assistance declarations, and the training-app
502-- fingerprint on fair-play reports.
503--
504-- positions: one row per position (FEN without the move counters), the
505-- deepest engine evaluation stored for it, filled by the fair-play worker
506-- as it screens games. A lookup is a key read.
507CREATE TABLE chess.positions (
508    key text PRIMARY KEY,
509    best text,
510    score_cp integer NOT NULL,
511    depth integer NOT NULL,
512    engine text NOT NULL,
513    seen integer NOT NULL DEFAULT 1,
514    updated_at timestamptz NOT NULL DEFAULT now());
515-- assist_marks: a player switching the training app's engine help on or
516-- off during one of our games, as declared by the app. Ground truth for
517-- training the fair-play model; screening never relies on it.
518CREATE TABLE chess.assist_marks (
519    id bigserial PRIMARY KEY,
520    game_id text NOT NULL REFERENCES chess.games ON DELETE CASCADE,
521    player_id bigint NOT NULL REFERENCES chess.accounts ON DELETE CASCADE,
522    ply integer NOT NULL,
523    assist boolean NOT NULL,
524    at timestamptz NOT NULL DEFAULT now());
525CREATE INDEX assist_marks_game_idx ON chess.assist_marks (game_id, player_id);
526ALTER TABLE chess.fairplay_reports
527    ADD COLUMN app_disagree integer,
528    ADD COLUMN app_picked integer,
529    ADD COLUMN app_picked_reference integer,
530    ADD COLUMN declared_assist boolean NOT NULL DEFAULT false;
531",
532    r"
533-- 3. Board photos that couldn't be read for sure, kept for review.
534--
535-- reading: what our board reader made of it (placement, confidence, unsure
536-- squares, model); suggestion: the vision language model's reading, which
537-- the Review page pre-fills; label: the position a person confirmed (FEN
538-- placement, White's side), by the player who sent it in the app or by a
539-- reviewer on the Review page. Labelled photos are training data.
540CREATE TABLE chess.board_photos (
541    id bigserial PRIMARY KEY,
542    player_id bigint,
543    created_at timestamptz NOT NULL DEFAULT now(),
544    image bytea NOT NULL,
545    mime text NOT NULL,
546    reason text NOT NULL CHECK (reason IN ('unsure', 'implausible', 'no_model')),
547    reading jsonb,
548    suggestion jsonb,
549    suggestion_failed text,
550    label text,
551    label_source text CHECK (label_source IN ('player', 'reviewer')),
552    unreadable boolean NOT NULL DEFAULT false,
553    labelled_by bigint,
554    labelled_at timestamptz);
555CREATE INDEX board_photos_open_idx ON chess.board_photos (id DESC) WHERE label IS NULL AND NOT unreadable;
556",
557    r"
558-- 4. Endgame tablebase answers (up to seven pieces), kept once fetched:
559-- the tablebase part of the position dictionary.
560CREATE TABLE chess.tablebase (
561    key text PRIMARY KEY,
562    answer jsonb NOT NULL,
563    fetched_at timestamptz NOT NULL DEFAULT now());
564",
565    r"
566-- 5. The position dictionary grows.
567--
568-- positions: positions people confirmed from board photos join it before
569-- anyone has evaluated them (score_cp NULL, photos counting them); the
570-- fair-play worker evaluates those when it has no game to screen, scoring
571-- every legal move (moves: [{uci, cp, depth}], from the side to move's
572-- view). eval_lease_until: a worker is on it.
573-- position_hints: analyses the training app sends in, one per position and
574-- player (the deepest kept). Hints, not trusted like the worker's.
575ALTER TABLE chess.positions ALTER COLUMN score_cp DROP NOT NULL;
576ALTER TABLE chess.positions ALTER COLUMN depth SET DEFAULT 0;
577ALTER TABLE chess.positions ALTER COLUMN engine DROP NOT NULL;
578ALTER TABLE chess.positions ADD COLUMN photos integer NOT NULL DEFAULT 0;
579ALTER TABLE chess.positions ADD COLUMN moves jsonb;
580ALTER TABLE chess.positions ADD COLUMN eval_lease_until timestamptz;
581CREATE INDEX positions_unscored_idx ON chess.positions (updated_at) WHERE score_cp IS NULL;
582CREATE TABLE chess.position_hints (
583    key text NOT NULL,
584    player_id bigint NOT NULL,
585    best text,
586    score_cp integer NOT NULL,
587    depth integer NOT NULL,
588    engine text NOT NULL,
589    moves jsonb,
590    at timestamptz NOT NULL DEFAULT now(),
591    PRIMARY KEY (key, player_id));
592",
593];
594
595#[cfg(test)]
596mod tests {
597    use super::*;
598
599    #[test]
600    fn neon_direct_host() {
601        assert_eq!(
602            direct_host("ep-cool-1234-pooler.eu-central-1.aws.neon.tech").as_deref(),
603            Some("ep-cool-1234.eu-central-1.aws.neon.tech")
604        );
605        assert_eq!(direct_host("localhost"), None);
606        assert_eq!(direct_host("db.internal"), None);
607    }
608
609    #[test]
610    fn listen_url_replaces_the_pooler_host() {
611        let u = listen_url(
612            "postgresql://u:p@ep-cool-1234-pooler.eu-central-1.aws.neon.tech/neondb?sslmode=require",
613        );
614        assert_eq!(u, "postgresql://u:p@ep-cool-1234.eu-central-1.aws.neon.tech/neondb?sslmode=require");
615        let pg = tokio_postgres::Config::from_str(&u).unwrap();
616        assert_eq!(pg.get_hosts().len(), 1, "one host: the direct one");
617        let local = "postgres://dev:dev@127.0.0.1/chess?sslmode=disable";
618        assert_eq!(listen_url(local), local);
619        assert_eq!(listen_url("host=x user=y"), "host=x user=y");
620    }
621
622    #[test]
623    fn notes_round_trip() {
624        for n in [
625            Note::Game("abcd1234".into()),
626            Note::Chat { game: "g".into(), line: serde_json::json!({"type": "chatLine"}) },
627            Note::User { player: 7, event: serde_json::json!({"type": "gameStart"}) },
628            Note::Resync,
629        ] {
630            let back = Db::decode(&Db::encode(&n)).unwrap();
631            assert_eq!(format!("{back:?}"), format!("{n:?}"));
632        }
633        assert!(Db::decode("nope").is_none());
634    }
635
636    #[test]
637    fn prefix_replacement_only_touches_table_names() {
638        let q = "SELECT * FROM chess.games g JOIN site.players p ON p.id = g.white_id -- the website. chess.";
639        assert_eq!(
640            prefix_replace(&prefix_replace(q, "chess.", "\"c\"."), "site.", "\"s\"."),
641            "SELECT * FROM \"c\".games g JOIN \"s\".players p ON p.id = g.white_id -- the website. chess."
642        );
643        assert_eq!(prefix_replace("(chess.a,chess.b)", "chess.", "x."), "(x.a,x.b)");
644    }
645
646    #[test]
647    fn schema_prefixes() {
648        assert_eq!(quote_ident("a\"b"), "\"a\"\"b\"");
649        assert!(MIGRATIONS[0].contains("REFERENCES site.players"));
650    }
651}
open_chess_server/db.rs

open_chess_server/
db.rs
