From e659a1dbd17c47f363e787b608f01f027cefc2bc Mon Sep 17 00:00:00 2001 From: Isaac Corbrey Date: Sun, 20 Sep 2026 08:21:08 -0400 Subject: [PATCH] fix: AT Proto repo as source of truth, multi-difficulty game pool Source-of-truth principle: - On startup, hydrate game state from network records (not SQLite) - SQLite is now purely a write-through cache, never read on startup - Spawn background reconciliation task (60s interval) to catch any drift from Jetstream reconnection gaps - reconcile_games re-fetches moves for all active games from network Multi-difficulty game pool: - ensure_waiting_games creates one waiting game per difficulty (1-5) - When a game gets a playerTwo, it's consumed from the pool - After each bot move, the pool is replenished - Challengers pick difficulty by joining the corresponding game Also: - MoveRecord deduplicated across main.rs and jetstream.rs - Fixed doc comment (not hardcoded 9x9) --- src/board.rs | 2 +- src/db.rs | 1 + src/jetstream.rs | 14 +-- src/main.rs | 271 ++++++++++++++++++++++++++++++++++++++++------- 4 files changed, 235 insertions(+), 53 deletions(-) diff --git a/src/board.rs b/src/board.rs index 172b828..dd1479c 100644 --- a/src/board.rs +++ b/src/board.rs @@ -1,6 +1,6 @@ use std::collections::HashSet; -/// 9×9 Go board. Cells are `u8` (0=empty, 1=black, 2=white) for compact +/// Go board. Cells are `u8` (0=empty, 1=black, 2=white) for compact /// storage and fast flood-fill — the domain-level `Color` enum lives in /// `crate::Color` and converts via `Color::as_u8` / `Color::from_u8`. #[derive(Clone, Debug)] diff --git a/src/db.rs b/src/db.rs index 137420f..3bf6102 100644 --- a/src/db.rs +++ b/src/db.rs @@ -14,6 +14,7 @@ impl Database { Self { pool } } + #[allow(dead_code)] pub async fn load_active_games(&self) -> Result> { let rows = sqlx::query( r#" diff --git a/src/jetstream.rs b/src/jetstream.rs index 13a3c1c..3abcf1b 100644 --- a/src/jetstream.rs +++ b/src/jetstream.rs @@ -6,7 +6,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info}; use crate::board::Board; -use crate::{unknown_to_value, App, Color, GameState, LastMove}; +use crate::{unknown_to_value, App, Color, GameState, LastMove, MoveRecord}; struct GameHandler { app: App, @@ -140,13 +140,6 @@ async fn handle_game( Ok(()) } -struct MoveRecord { - x: usize, - y: usize, - color: Color, - move_number: u32, -} - async fn fetch_game_moves(app: &App, repo: &str, _game_rkey: &str) -> Result> { let game_uri_prefix = format!("at://{}/boo.sky.go.game/", repo); @@ -355,7 +348,7 @@ async fn post_move(app: &App, game: &GameState, x: usize, y: usize) -> Result<() } info!("Move posted successfully: {} at {},{}", color_name, x, y); - crate::ensure_waiting_game(app).await?; + crate::ensure_waiting_games(app).await?; Ok(()) } @@ -396,7 +389,7 @@ async fn resign(app: &App, game: &GameState) -> Result<()> { app.db.upsert_game(game).await?; } - crate::ensure_waiting_game(app).await?; + crate::ensure_waiting_games(app).await?; Ok(()) } @@ -436,7 +429,6 @@ pub async fn run_event_loop(app: &App) -> Result<()> { Ok(()) } -#[allow(dead_code)] pub async fn reconcile_games(app: &App) -> Result<()> { let game_uris: Vec = { let games = app.games.lock().await; diff --git a/src/main.rs b/src/main.rs index 96e337a..26055e0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -105,6 +105,170 @@ fn unknown_to_value(u: &atrium_api::types::Unknown) -> serde_json::Value { serde_json::to_value(u).unwrap_or(serde_json::Value::Null) } +/// Hydrate game state from AT Proto repo (source of truth) on startup. +/// Fetches all game records, finds active ones where bot is playerTwo, +/// replays moves to rebuild board state, then persists to SQLite cache. +async fn hydrate_from_network(app: &App) -> Result<()> { + let did = app.agent.did().await.expect("authenticated"); + let did_str = did.as_str(); + + // Fetch all game records from bot's repo + let games_resp = app + .agent + .api + .com + .atproto + .repo + .list_records( + atrium_api::com::atproto::repo::list_records::ParametersData { + collection: "boo.sky.go.game".parse().unwrap(), + cursor: None, + limit: Some(100u8.try_into().unwrap()), + repo: did.as_str().parse().unwrap(), + reverse: None, + } + .into(), + ) + .await?; + + // Fetch all move records from bot's repo + let moves_resp = app + .agent + .api + .com + .atproto + .repo + .list_records( + atrium_api::com::atproto::repo::list_records::ParametersData { + collection: "boo.sky.go.move".parse().unwrap(), + cursor: None, + limit: Some(100u8.try_into().unwrap()), + repo: did.as_str().parse().unwrap(), + reverse: None, + } + .into(), + ) + .await?; + + // Index moves by game URI + let mut moves_by_game: std::collections::HashMap> = + std::collections::HashMap::new(); + for r in &moves_resp.records { + let val = unknown_to_value(&r.value); + let map = match val.as_object() { + Some(m) => m, + None => continue, + }; + let game_ref = match map.get("game").and_then(|v| v.as_str()) { + Some(g) => g.to_string(), + None => continue, + }; + let color = match map.get("color").and_then(|v| v.as_str()) { + Some("black") => Color::Black, + _ => Color::White, + }; + moves_by_game + .entry(game_ref) + .or_default() + .push(MoveRecord { + x: map.get("x").and_then(|v| v.as_u64()).unwrap_or(0) as usize, + y: map.get("y").and_then(|v| v.as_u64()).unwrap_or(0) as usize, + color, + move_number: map.get("moveNumber").and_then(|v| v.as_u64()).unwrap_or(0) as u32, + }); + } + + // For each game record, check if bot is involved and game is active + let mut games_map = app.games.lock().await; + for record in &games_resp.records { + let val = unknown_to_value(&record.value); + let map = match val.as_object() { + Some(m) => m, + None => continue, + }; + + let status = map.get("status").and_then(|v| v.as_str()).unwrap_or(""); + if status != "active" { + continue; + } + + let player_two = match map.get("playerTwo").and_then(|v| v.as_str()) { + Some(p) => p, + None => continue, + }; + + // Bot must be one of the players + let player_one = map.get("playerOne").and_then(|v| v.as_str()).unwrap_or(""); + let bot_is_p1 = player_one == did_str; + let bot_is_p2 = player_two == did_str; + if !bot_is_p1 && !bot_is_p2 { + continue; + } + + let game_uri = record.uri.to_string(); + let board_size = map.get("boardSize").and_then(|v| v.as_u64()).unwrap_or(9) as usize; + let difficulty = map.get("difficulty").and_then(|v| v.as_u64()).unwrap_or(1) as u8; + let bot_color = if bot_is_p1 { + Color::Black + } else { + Color::White + }; + + // Replay moves from network (source of truth) + let mut board = crate::board::Board::new(board_size); + let mut move_number: u32 = 0; + let mut last_move = None; + + if let Some(moves) = moves_by_game.get(&game_uri) { + let mut sorted_moves = moves.clone(); + sorted_moves.sort_by_key(|m| m.move_number); + for m in &sorted_moves { + let caps = board.apply_move(m.x, m.y, m.color.as_u8()); + move_number = m.move_number; + last_move = Some(LastMove { + x: m.x, + y: m.y, + color: m.color, + captures: caps, + }); + } + } + + let game_state = GameState { + game_uri: game_uri.clone(), + board_size, + bot_color, + difficulty, + board, + move_number, + status: "active".into(), + player_one: player_one.to_string(), + player_two: player_two.to_string(), + last_move, + }; + + info!( + "Hydrated from network: {} (move {}, difficulty {})", + game_uri, move_number, difficulty + ); + + // Persist to SQLite cache and in-memory map + app.db.upsert_game(&game_state).await?; + games_map.insert(game_uri.clone(), game_state); + } + + Ok(()) +} + +// MoveRecord is shared between hydration and jetstream +#[derive(Clone)] +pub struct MoveRecord { + pub x: usize, + pub y: usize, + pub color: Color, + pub move_number: u32, +} + async fn authenticate(config: &Config) -> Result { let pds_endpoint = if config.identifier.contains(':') { let http = reqwest::Client::new(); @@ -142,7 +306,10 @@ async fn authenticate(config: &Config) -> Result { Ok(agent) } -async fn ensure_waiting_game(app: &App) -> Result { +/// Ensure one waiting game per difficulty level (1–5). +/// Each game has `playerOne: bot_did` and `status: "waiting"`. +/// When a game gets a `playerTwo`, it leaves the pool and gets replenished. +async fn ensure_waiting_games(app: &App) -> Result<()> { let did = app.agent.did().await.expect("authenticated"); let resp = app @@ -163,49 +330,59 @@ async fn ensure_waiting_game(app: &App) -> Result { ) .await?; + // Find which difficulties already have a waiting game + let mut covered_difficulties = std::collections::HashSet::new(); for record in &resp.records { let val = unknown_to_value(&record.value); if let Some(obj) = val.as_object() { let status = obj.get("status").and_then(|v| v.as_str()); let p1 = obj.get("playerOne").and_then(|v| v.as_str()); + let diff = obj.get("difficulty").and_then(|v| v.as_u64()).unwrap_or(1) as u8; if status == Some("waiting") && p1 == Some(did.as_str()) { - info!("Found existing waiting game: {}", record.uri); - return Ok(record.uri.to_string()); + covered_difficulties.insert(diff); } } } - let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); - let record = serde_json::json!({ - "$type": "boo.sky.go.game", - "status": "waiting", - "boardSize": 9, - "createdAt": now, - "playerOne": did.as_str(), - "difficulty": 1 - }); + // Create missing waiting games for each difficulty level + for difficulty in 1u8..=5 { + if covered_difficulties.contains(&difficulty) { + continue; + } - let resp = app - .agent - .api - .com - .atproto - .repo - .create_record( - atrium_api::com::atproto::repo::create_record::InputData { - collection: "boo.sky.go.game".parse().unwrap(), - record: serde_json::from_value(record).unwrap(), - repo: did.clone().into(), - rkey: None, - swap_commit: None, - validate: None, - } - .into(), - ) - .await?; + let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let record = serde_json::json!({ + "$type": "boo.sky.go.game", + "status": "waiting", + "boardSize": 9, + "createdAt": now, + "playerOne": did.as_str(), + "difficulty": difficulty, + }); + + let resp = app + .agent + .api + .com + .atproto + .repo + .create_record( + atrium_api::com::atproto::repo::create_record::InputData { + collection: "boo.sky.go.game".parse().unwrap(), + record: serde_json::from_value(record).unwrap(), + repo: did.as_str().parse().unwrap(), + rkey: None, + swap_commit: None, + validate: None, + } + .into(), + ) + .await?; + + info!("Created waiting game (difficulty {}): {}", difficulty, resp.uri); + } - info!("Created waiting game: {}", resp.uri); - Ok(resp.uri.to_string()) + Ok(()) } #[tokio::main] @@ -229,21 +406,33 @@ async fn main() -> Result<()> { db::run_migrations(&pool).await?; let db = db::Database::new(pool); - let active_games = db.load_active_games().await?; - let mut games_map = HashMap::new(); - for gs in active_games { - info!("Hydrated game {} (move {})", gs.game_uri, gs.move_number); - games_map.insert(gs.game_uri.clone(), gs); - } - + // Start with empty map — hydrate from AT Proto repo (source of truth) let app = App(Arc::new(AppInner { config: config.clone(), agent, db, - games: Mutex::new(games_map), + games: Mutex::new(HashMap::new()), })); - ensure_waiting_game(&app).await?; + // Hydrate from network — AT Proto repo is truth, SQLite is cache + hydrate_from_network(&app).await?; + + // Ensure one waiting game per difficulty level + ensure_waiting_games(&app).await?; + + // Spawn periodic reconciliation to catch any drift from Jetstream reconnection + let reconciliation_app = app.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(60)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + if let Err(e) = jetstream::reconcile_games(&reconciliation_app).await { + tracing::error!("Reconciliation failed: {:#}", e); + } + } + }); + jetstream::run_event_loop(&app).await?; Ok(()) -- knotmirror 0.0.1