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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731mod board;mod db;mod jetstream;mod jev;mod state;
use anyhow::{Context, Result};use atrium_api::agent::atp_agent::{store::MemorySessionStore, AtpAgent};use atrium_api::com::atproto::repo::get_record::Error as GetRecordError;use atrium_api::xrpc::error::{Error as XrpcError, XrpcErrorKind};use atrium_xrpc_client::reqwest::ReqwestClient;use sqlx::sqlite::SqlitePoolOptions;use std::collections::HashMap;use std::sync::Arc;use tokio::sync::Mutex;use tracing::info;
#[derive(Clone)]struct Config { identifier: String, password: String, typesafe_api_key: String, database_url: String, jetstream_url: String,}
impl Config { fn from_env() -> Result<Self> { Ok(Self { identifier: std::env::var("ATPROTO_IDENTIFIER").context("ATPROTO_IDENTIFIER")?, password: std::env::var("ATPROTO_PASSWORD").context("ATPROTO_PASSWORD")?, typesafe_api_key: std::env::var("TYPESAFE_API_KEY").context("TYPESAFE_API_KEY")?, database_url: std::env::var("DATABASE_URL") .unwrap_or_else(|_| "sqlite://./jevgo.db".into()), jetstream_url: std::env::var("JETSTREAM_URL") .unwrap_or_else(|_| "wss://jetstream2.us-east.bsky.network/subscribe".into()), }) }}
type Agent = AtpAgent<MemorySessionStore, ReqwestClient>;
struct AppInner { config: Config, agent: Agent, db: db::Database, games: Mutex<HashMap<String, GameState>>,}
/// Cloneable app handle — `AtpAgent` doesn't impl Clone, so we wrap in `Arc`.#[derive(Clone)]struct App(Arc<AppInner>);
impl std::ops::Deref for App { type Target = AppInner; fn deref(&self) -> &Self::Target { &self.0 }}
#[derive(Clone, Debug)]pub struct GameState { pub game_uri: String, pub board_size: usize, pub bot_color: Color, pub board: board::Board, pub move_number: u32, pub status: String, pub player_one: String, pub player_two: String, pub last_move: Option<LastMove>, /// Consecutive passes ending at the current move_number. Two in a row /// (regardless of who) means the game is over and needs scoring. pub passes_in_a_row: u8,}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]pub enum Color { Black, White,}
impl Color { pub fn as_u8(self) -> u8 { match self { Color::Black => 1, Color::White => 2, } }
pub fn from_u8(v: u8) -> Self { match v { 1 => Color::Black, _ => Color::White, } }
pub fn name(self) -> &'static str { match self { Color::Black => "black", Color::White => "white", } }}
#[derive(Clone, Debug)]pub struct LastMove { pub x: usize, pub y: usize, pub color: Color, pub captures: usize,}
/// Convert atrium's `Unknown` (opaque JSON) to `serde_json::Value`.fn unknown_to_value(u: &atrium_api::types::Unknown) -> serde_json::Value { serde_json::to_value(u).unwrap_or(serde_json::Value::Null)}
/// Resolve a DID's own PDS service endpoint (`did:plc:` via the PLC/// directory, `did:web:` via its domain's well-known document).async fn resolve_pds(did: &str) -> Result<String> { let http = reqwest::Client::new(); let doc: serde_json::Value = if let Some(domain) = did.strip_prefix("did:web:") { http.get(format!("https://{}/.well-known/did.json", domain)) .send() .await? .json() .await? } else { http.get(format!("https://plc.directory/{}", did)) .send() .await? .json() .await? };
doc["service"] .as_array() .and_then(|svcs| { svcs.iter() .find(|s| s["type"].as_str() == Some("AtprotoPersonalDataServer")) }) .and_then(|s| s["serviceEndpoint"].as_str()) .map(str::to_string) .ok_or_else(|| anyhow::anyhow!("no AtprotoPersonalDataServer service for {}", did))}
/// List every record in `collection` for `repo` as (AT URI, raw JSON value)/// pairs, following pagination to the end. `com.atproto.repo.listRecords` is/// scoped to whichever PDS serves it — the bot's own session only reaches/// its own repo, so any other repo needs its own PDS resolved and queried/// directly (an unauthenticated read, same as any public record).async fn list_repo_records( app: &App, repo: &str, collection: &str,) -> Result<Vec<(String, serde_json::Value)>> { let bot_did = app.agent.did().await.expect("authenticated"); let mut records = Vec::new();
if repo == bot_did.as_str() { let mut cursor = None; loop { let resp = app .agent .api .com .atproto .repo .list_records( atrium_api::com::atproto::repo::list_records::ParametersData { collection: collection.parse().unwrap(), cursor: cursor.clone(), limit: Some(100u8.try_into().unwrap()), repo: repo.parse().unwrap(), reverse: None, } .into(), ) .await?; records.extend( resp.records .iter() .map(|r| (r.uri.to_string(), unknown_to_value(&r.value))), ); // A cursor that repeats itself is a broken server; stop rather // than page forever. if resp.cursor.is_none() || resp.cursor == cursor { break; } cursor = resp.cursor.clone(); } } else { let pds = resolve_pds(repo).await?; let mut cursor: Option<String> = None; loop { let mut query = vec![("repo", repo), ("collection", collection), ("limit", "100")]; if let Some(c) = &cursor { query.push(("cursor", c.as_str())); } let resp: serde_json::Value = reqwest::Client::new() .get(format!("{}/xrpc/com.atproto.repo.listRecords", pds)) .query(&query) .send() .await? .json() .await?; let page = resp["records"].as_array().cloned().unwrap_or_default(); records.extend(page.into_iter().filter_map(|r| { Some((r["uri"].as_str()?.to_string(), r["value"].clone())) }));
let next = resp["cursor"].as_str().map(str::to_string); if next.is_none() || next == cursor { break; } cursor = next; } }
Ok(records)}
/// Whether an XRPC error body means "no such record". A deleted record is/// a stale reference — normal noise in replayed event streams — while every/// other failure deserves the loud error path.fn xrpc_kind_is_not_found(kind: &XrpcErrorKind<GetRecordError>) -> bool { match kind { XrpcErrorKind::Custom(GetRecordError::RecordNotFound(_)) => true, XrpcErrorKind::Undefined(body) => body.error.as_deref() == Some("RecordNotFound"), _ => false, }}
/// Whether a get_record failure means "no such record".pub(crate) fn record_not_found(err: &XrpcError<GetRecordError>) -> bool { match err { XrpcError::XrpcResponse(xe) => xe.error.as_ref().is_some_and(xrpc_kind_is_not_found), _ => false, }}
/// Fetch a single record by AT URI from whichever repo holds it, as raw/// JSON. `Ok(None)` means the record does not exist (deleted or never/// written) — callers treat that as stale input, not an error.async fn fetch_game_record(app: &App, record_uri: &str) -> Result<Option<serde_json::Value>> { let rest = record_uri .strip_prefix("at://") .with_context(|| format!("not an at:// URI: {record_uri}"))?; let (repo, tail) = rest.split_once('/').context("malformed at:// URI")?; let (collection, rkey) = tail.split_once('/').context("malformed at:// URI")?;
let bot_did = app.agent.did().await.expect("authenticated"); if repo == bot_did.as_str() { match app .agent .api .com .atproto .repo .get_record( atrium_api::com::atproto::repo::get_record::ParametersData { cid: None, collection: collection.parse().unwrap(), repo: repo.parse().unwrap(), rkey: rkey.parse().unwrap(), } .into(), ) .await { Ok(record) => Ok(Some(unknown_to_value(&record.value))), Err(e) if record_not_found(&e) => Ok(None), Err(e) => Err(e.into()), } } else { let pds = resolve_pds(repo).await?; let resp = reqwest::Client::new() .get(format!("{}/xrpc/com.atproto.repo.getRecord", pds)) .query(&[("repo", repo), ("collection", collection), ("rkey", rkey)]) .send() .await?; if !resp.status().is_success() { tracing::warn!(uri = %record_uri, status = %resp.status(), "record not fetchable"); return Ok(None); } let body: serde_json::Value = resp.json().await?; Ok(Some(body["value"].clone())) }}
#[cfg(test)]mod error_tests { use super::*; use atrium_api::xrpc::error::ErrorResponseBody;
#[test] fn record_not_found_matches_only_missing_records() { let missing: XrpcErrorKind<GetRecordError> = XrpcErrorKind::Custom(GetRecordError::RecordNotFound(None)); assert!(xrpc_kind_is_not_found(&missing));
let undefined_name: XrpcErrorKind<GetRecordError> = XrpcErrorKind::Undefined(ErrorResponseBody { error: Some("RecordNotFound".into()), message: Some("Could not find record".into()), }); assert!(xrpc_kind_is_not_found(&undefined_name));
let other: XrpcErrorKind<GetRecordError> = XrpcErrorKind::Undefined(ErrorResponseBody { error: Some("InvalidRequest".into()), message: None, }); assert!(!xrpc_kind_is_not_found(&other), "real failures must stay loud"); }}
/// Fetch a single game's moves and passes posted into `repo` (one player's/// own repo), filtered by exact `game` URI match. Both share one/// moveNumber sequence (per the boo.sky.go.pass lexicon), so they're/// merged into one ply list here.async fn fetch_plies_from_repo(app: &App, repo: &str, game_uri: &str) -> Result<Vec<MoveRecord>> { let mut plies = Vec::new();
for (_, val) in list_repo_records(app, repo, "boo.sky.go.move").await? { let Some(map) = val.as_object() else { continue }; if map.get("game").and_then(|v| v.as_str()) != Some(game_uri) { continue; } let (Some(x), Some(y), Some(move_number)) = ( map.get("x").and_then(|v| v.as_u64()), map.get("y").and_then(|v| v.as_u64()), map.get("moveNumber").and_then(|v| v.as_u64()), ) else { continue; }; let color = match map.get("color").and_then(|v| v.as_str()) { Some("black") => Color::Black, _ => Color::White, }; plies.push(MoveRecord { x: x as usize, y: y as usize, color, move_number: move_number as u32, is_pass: false }); }
for (_, val) in list_repo_records(app, repo, "boo.sky.go.pass").await? { let Some(map) = val.as_object() else { continue }; if map.get("game").and_then(|v| v.as_str()) != Some(game_uri) { continue; } let Some(move_number) = map.get("moveNumber").and_then(|v| v.as_u64()) else { continue; }; let color = match map.get("color").and_then(|v| v.as_str()) { Some("black") => Color::Black, _ => Color::White, }; plies.push(MoveRecord { x: 0, y: 0, color, move_number: move_number as u32, is_pass: true }); }
Ok(plies)}
// MoveRecord is shared between hydration and jetstream. Represents either a// stone placement or a pass — both consume a moveNumber in the same// sequence (per the boo.sky.go.pass lexicon), so they replay together.#[derive(Clone)]pub struct MoveRecord { pub x: usize, pub y: usize, pub color: Color, pub move_number: u32, pub is_pass: bool,}
/// Replay moves and passes onto `board` in `move_number` order. Skips (and/// logs) any placement that lands on an already-occupied cell instead of/// panicking — the on-chain history is untrusted input, and a/// duplicate/corrupt record must not take down the whole bot process./// Returns the final move number, the last stone placed (if any), and how/// many passes in a row ended the sequence — two means the game is over.fn replay_moves(board: &mut board::Board, moves: &[MoveRecord]) -> (u32, Option<LastMove>, u8) { let mut sorted = moves.to_vec(); sorted.sort_by_key(|m| m.move_number);
let mut move_number = 0; let mut last_move = None; let mut passes_in_a_row = 0u8; for m in &sorted { if m.is_pass { move_number = m.move_number; passes_in_a_row = passes_in_a_row.saturating_add(1); continue; } match board.try_apply_move(m.x, m.y, m.color.as_u8()) { Some(captures) => { move_number = m.move_number; passes_in_a_row = 0; last_move = Some(LastMove { x: m.x, y: m.y, color: m.color, captures, }); } None => { tracing::error!( x = m.x, y = m.y, move_number = m.move_number, "skipping corrupt move: lands on an occupied cell" ); } } } (move_number, last_move, passes_in_a_row)}
#[cfg(test)]mod replay_tests { use super::*;
fn mv(x: usize, y: usize, color: Color, move_number: u32) -> MoveRecord { MoveRecord { x, y, color, move_number, is_pass: false } }
fn pass(color: Color, move_number: u32) -> MoveRecord { MoveRecord { x: 0, y: 0, color, move_number, is_pass: true } }
#[test] fn replay_skips_duplicate_move_number_instead_of_panicking() { let moves = vec![ mv(4, 4, Color::Black, 1), mv(3, 3, Color::Black, 3), mv(3, 3, Color::Black, 3), // duplicate: same move_number and cell mv(4, 3, Color::Black, 5), mv(2, 3, Color::Black, 7), mv(3, 5, Color::Black, 9), mv(3, 5, Color::Black, 9), // duplicate: same move_number and cell mv(3, 6, Color::Black, 11), mv(5, 4, Color::Black, 13), ];
let mut board = board::Board::new(9); let (move_number, last_move, passes) = replay_moves(&mut board, &moves);
assert_eq!(move_number, 13); assert_eq!(passes, 0); assert_eq!(board.cells[3][3], Color::Black.as_u8()); // (3,3): cells[y][x] assert_eq!(board.cells[5][3], Color::Black.as_u8()); // (3,5) let last = last_move.expect("a move was applied"); assert_eq!((last.x, last.y), (5, 4));
let stones_on_board = board .cells .iter() .flatten() .filter(|&&c| c != 0) .count(); assert_eq!(stones_on_board, moves.len() - 2, "duplicates must not double-place stones"); }
/// Reproduces the actual stuck state: before the cross-repo fetch fix, /// hydration only ever saw the bot's own (black) moves for this real /// game, landing on move 13 and waiting forever for a white reply that /// had already been posted — to the opponent's own repo, which the bot /// never queried. Merging both players' real move records must land on /// move 14 (the opponent's actual latest move). #[test] fn replay_merges_both_players_real_moves_past_the_stuck_point() { let black_moves = vec![ mv(4, 4, Color::Black, 1), mv(3, 3, Color::Black, 3), mv(4, 3, Color::Black, 5), mv(2, 3, Color::Black, 7), mv(3, 5, Color::Black, 9), mv(3, 6, Color::Black, 11), mv(5, 4, Color::Black, 13), ]; let white_moves = vec![ mv(2, 6, Color::White, 2), mv(3, 4, Color::White, 4), mv(2, 4, Color::White, 6), mv(1, 4, Color::White, 8), mv(2, 5, Color::White, 10), mv(2, 7, Color::White, 12), mv(4, 6, Color::White, 14), ];
let mut bot_only = board::Board::new(9); let (bot_only_move_number, _, _) = replay_moves(&mut bot_only, &black_moves); assert_eq!(bot_only_move_number, 13, "black-only replay is the stuck state");
let mut merged = black_moves; merged.extend(white_moves); let mut board = board::Board::new(9); let (move_number, last_move, passes) = replay_moves(&mut board, &merged);
assert_eq!(move_number, 14, "merging both repos' moves must reach white's latest reply"); assert_eq!(passes, 0); let last = last_move.expect("a move was applied"); assert_eq!((last.x, last.y, last.color), (4, 6, Color::White)); }
#[test] fn replay_tracks_two_consecutive_passes() { let moves = vec![ mv(4, 4, Color::Black, 1), mv(2, 6, Color::White, 2), pass(Color::Black, 3), pass(Color::White, 4), ];
let mut board = board::Board::new(9); let (move_number, last_move, passes) = replay_moves(&mut board, &moves);
assert_eq!(move_number, 4); assert_eq!(passes, 2, "two consecutive passes must be tracked"); // last_move stays the last real placement — passes don't have coordinates let last = last_move.expect("a stone was placed before the passes"); assert_eq!((last.x, last.y, last.color), (2, 6, Color::White)); }
#[test] fn replay_resets_pass_streak_on_real_move() { let moves = vec![ mv(4, 4, Color::Black, 1), pass(Color::White, 2), mv(3, 3, Color::Black, 3), ];
let mut board = board::Board::new(9); let (move_number, _, passes) = replay_moves(&mut board, &moves);
assert_eq!(move_number, 3); assert_eq!(passes, 0, "a real move after a pass resets the streak"); }}
async fn authenticate(config: &Config) -> Result<Agent> { let pds_endpoint = if config.identifier.contains(':') { let http = reqwest::Client::new(); let doc: serde_json::Value = http .get(format!("https://plc.directory/{}", config.identifier)) .send() .await? .json() .await?; doc["service"] .as_array() .and_then(|svcs| { svcs.iter() .find(|s| s["type"].as_str() == Some("AtprotoPersonalDataServer")) }) .and_then(|s| s["serviceEndpoint"].as_str()) .unwrap_or("https://bsky.social") .to_string() } else { "https://bsky.social".into() };
let client = ReqwestClient::new(&pds_endpoint); let agent = AtpAgent::new(client, MemorySessionStore::default()); agent .login(&config.identifier, &config.password) .await .context("AT Protocol login failed")?;
let did = agent.did().await.expect("login sets DID"); info!(pds = %pds_endpoint, did = %did.as_str(), "authenticated");
Ok(agent)}
const BOARD_SIZES: [u64; 3] = [9, 13, 19];
/// Ensure one open waiting game per board size (9x9, 13x13, 19x19). Every/// size gets the same full state — no weaker tiers, just more board to/// play on. `playerOne: bot_did`, `status: "waiting"`; when one 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 .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?;
let mut covered_sizes = 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 size = obj.get("boardSize").and_then(|v| v.as_u64()).unwrap_or(9); if status == Some("waiting") && p1 == Some(did.as_str()) { covered_sizes.insert(size); } } }
for board_size in BOARD_SIZES { if covered_sizes.contains(&board_size) { continue; }
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": board_size, "createdAt": now, "playerOne": did.as_str(), });
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!(board_size, game = %resp.uri, "created waiting game"); }
Ok(())}
#[tokio::main]async fn main() -> Result<()> { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| "jevgo=info,atproto_jetstream=info".into()), ) .init();
let config = Config::from_env()?;
// Retry auth with exponential backoff — prevents crash-looping on bad creds // or transient PDS errors, which would hit rate limits with Railway's restart policy. let agent = { let mut backoff = std::time::Duration::from_secs(2); let max_backoff = std::time::Duration::from_secs(120); loop { match authenticate(&config).await { Ok(agent) => break agent, Err(e) => { tracing::error!( retry_in_secs = backoff.as_secs_f64(), error = %e, "auth failed, retrying" ); tokio::time::sleep(backoff).await; backoff = (backoff * 2).min(max_backoff); } } } };
use sqlx::sqlite::SqliteConnectOptions; use std::str::FromStr;
let pool = SqlitePoolOptions::new() .max_connections(5) .connect_with( SqliteConnectOptions::from_str(&config.database_url)? .create_if_missing(true), ) .await .context("Failed to connect to SQLite")?;
db::run_migrations(&pool).await?; let db = db::Database::new(pool);
// 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(HashMap::new()), }));
// Hydrate from network — AT Proto repos are truth, SQLite caches which // games we're in (the only handle on challenge games after a restart) jetstream::hydrate(&app).await?;
// Ensure there's an open waiting game per board size 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!(error = %e, "reconciliation failed"); } } });
jetstream::run_event_loop(&app).await?;
Ok(())}