Something went wrong. Try again.
A bot for https://go.sky.boo that lets you play Go against TypeSafe's Jev.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146use anyhow::Result;use sqlx::sqlite::SqlitePool;
use crate::GameState;
#[derive(Clone)]pub struct Database { pool: SqlitePool,}
impl Database { pub fn new(pool: SqlitePool) -> Self { Self { pool } }
pub async fn upsert_game(&self, game: &GameState) -> Result<()> { let board_state = serde_json::to_string(&game.board.cells)?; let ko_point = game .board .ko_point .map(|(x, y)| serde_json::to_string(&(x, y)).unwrap());
sqlx::query( r#" INSERT INTO games (game_uri, board_size, bot_color, board_state, ko_point, move_number, status) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(game_uri) DO UPDATE SET board_state = excluded.board_state, ko_point = excluded.ko_point, move_number = excluded.move_number, status = excluded.status "#, ) .bind(&game.game_uri) .bind(game.board_size as i64) .bind(game.bot_color.name()) .bind(&board_state) .bind(&ko_point) .bind(game.move_number as i64) .bind(&game.status) .execute(&self.pool) .await?;
Ok(()) }
/// The last jetstream position known to be fully processed, in /// microseconds since the epoch. pub async fn get_cursor(&self) -> Result<Option<i64>> { let row: Option<(i64,)> = sqlx::query_as("SELECT time_us FROM jetstream_cursor WHERE id = 0") .fetch_optional(&self.pool) .await?; Ok(row.map(|(time_us,)| time_us)) }
/// Checkpoint that every event up to `time_us` has been processed. pub async fn set_cursor(&self, time_us: i64) -> Result<()> { sqlx::query( r#" INSERT INTO jetstream_cursor (id, time_us) VALUES (0, ?) ON CONFLICT(id) DO UPDATE SET time_us = excluded.time_us "#, ) .bind(time_us) .execute(&self.pool) .await?;
Ok(()) }
/// Every game URI in the cache — the rediscovery handle for challenge /// games, which live in repos we can't list. pub async fn game_uris(&self) -> Result<Vec<String>> { let rows: Vec<(String,)> = sqlx::query_as("SELECT game_uri FROM games") .fetch_all(&self.pool) .await?; Ok(rows.into_iter().map(|(uri,)| uri).collect()) }
/// Drop a game from the cache once it needs no further attention. pub async fn forget_game(&self, game_uri: &str) -> Result<()> { sqlx::query("DELETE FROM games WHERE game_uri = ?") .bind(game_uri) .execute(&self.pool) .await?; Ok(()) }}
pub async fn run_migrations(pool: &SqlitePool) -> Result<()> { sqlx::query( r#" CREATE TABLE IF NOT EXISTS games ( game_uri TEXT PRIMARY KEY, board_size INTEGER NOT NULL DEFAULT 9, bot_color TEXT NOT NULL, board_state TEXT NOT NULL, ko_point TEXT, move_number INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'active' ) "#, ) .execute(pool) .await?;
sqlx::query( r#" CREATE TABLE IF NOT EXISTS jetstream_cursor ( id INTEGER PRIMARY KEY CHECK (id = 0), time_us INTEGER NOT NULL ) "#, ) .execute(pool) .await?;
Ok(())}
#[cfg(test)]mod tests { use super::*;
#[tokio::test] async fn cursor_checkpoints_and_advances() { let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); run_migrations(&pool).await.unwrap(); let db = Database::new(pool);
assert_eq!(db.get_cursor().await.unwrap(), None, "fresh database has no cursor");
db.set_cursor(1757517640675638).await.unwrap(); assert_eq!(db.get_cursor().await.unwrap(), Some(1757517640675638));
db.set_cursor(1757517640675999).await.unwrap(); assert_eq!( db.get_cursor().await.unwrap(), Some(1757517640675999), "a newer checkpoint must overwrite the old one" ); }}