refact: migrate sqlx + sqlite to tokio-postgresql

server/relay-server/src/db.rs

@@ -1,14 +1,14 @@
 //! Database access layer.
 //!
-//! All SQLite interaction is funnelled through this module. Functions return
-//! `sqlx::Result` so callers can handle errors uniformly.
+//! All PostgreSQL interaction is funnelled through this module. Functions return
+//! `Result<_, DbError>` so callers can handle errors uniformly.
 
-use sqlx::sqlite::SqliteConnectOptions;
-use sqlx::{SqlitePool, pool::PoolOptions};
+use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
+use tokio_postgres::{NoTls, error::SqlState};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 /// A registered user as stored in the database.
-#[derive(Clone, Debug, sqlx::FromRow)]
+#[derive(Clone, Debug)]
 pub struct User {
     pub id: i64,
     pub username: String,
@@ -18,7 +18,6 @@ pub struct User {
 }
 
 /// Aggregated game statistics for a user's public profile.
-#[derive(sqlx::FromRow)]
 pub struct UserStats {
     pub total: i64,
     pub wins: i64,
@@ -27,7 +26,6 @@ pub struct UserStats {
 }
 
 /// A condensed game entry returned by [`get_user_games`].
-#[derive(sqlx::FromRow)]
 pub struct GameSummary {
     pub id: i64,
     pub game_id: String,
@@ -38,6 +36,24 @@ pub struct GameSummary {
     pub outcome: Option<String>,
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum DbError {
+    #[error("connection pool error: {0}")]
+    Pool(#[from] deadpool_postgres::PoolError),
+    #[error("database error: {0}")]
+    Db(#[from] tokio_postgres::Error),
+}
+
+impl DbError {
+    pub fn is_unique_violation(&self) -> bool {
+        if let DbError::Db(e) = self {
+            e.code() == Some(&SqlState::UNIQUE_VIOLATION)
+        } else {
+            false
+        }
+    }
+}
+
 fn now_unix() -> i64 {
     SystemTime::now()
         .duration_since(UNIX_EPOCH)
@@ -45,34 +61,28 @@ fn now_unix() -> i64 {
         .as_secs() as i64
 }
 
-/// Opens (or creates) the SQLite database at `path` and runs all pending migrations.
-pub async fn init_db(path: &str) -> SqlitePool {
-    if let Some(parent) = std::path::Path::new(path).parent() {
-        if !parent.as_os_str().is_empty() {
-            tokio::fs::create_dir_all(parent)
-                .await
-                .expect("Failed to create database directory");
-        }
-    }
+/// Connects to the PostgreSQL database at `url` and runs all pending migrations.
+pub async fn init_db(url: &str) -> Pool {
+    let pg_config: tokio_postgres::Config = url.parse().expect("Invalid DATABASE_URL");
+    let manager = Manager::from_config(
+        pg_config,
+        NoTls,
+        ManagerConfig { recycling_method: RecyclingMethod::Fast },
+    );
+    let pool = Pool::builder(manager)
+        .max_size(5)
+        .build()
+        .expect("Failed to build connection pool");
 
-    let pool = PoolOptions::<sqlx::Sqlite>::new()
-        .max_connections(5)
-        .connect_with(
-            SqliteConnectOptions::new()
-                .filename(path)
-                .create_if_missing(true),
-        )
+    let client = pool.get().await.expect("Failed to get connection for migrations");
+    client
+        .batch_execute(include_str!("../migrations/001_init.sql"))
         .await
-        .expect("Failed to open SQLite database");
-
-    sqlx::migrate::Migrator::new(
-        std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/migrations")),
-    )
-    .await
-    .expect("Failed to locate migrations directory")
-    .run(&pool)
-    .await
-    .expect("Failed to run database migrations");
+        .expect("Migration 001 failed");
+    client
+        .batch_execute(include_str!("../migrations/002_participants_unique.sql"))
+        .await
+        .expect("Migration 002 failed");
 
     pool
 }
@@ -80,135 +90,164 @@ pub async fn init_db(path: &str) -> SqlitePool {
 // ── Users ────────────────────────────────────────────────────────────────────
 
 pub async fn create_user(
-    pool: &SqlitePool,
+    pool: &Pool,
     username: &str,
     email: &str,
     password_hash: &str,
-) -> sqlx::Result<i64> {
-    let id = sqlx::query(
-        "INSERT INTO users (username, email, password_hash, created_at) VALUES (?, ?, ?, ?)",
-    )
-    .bind(username)
-    .bind(email)
-    .bind(password_hash)
-    .bind(now_unix())
-    .execute(pool)
-    .await?
-    .last_insert_rowid();
-    Ok(id)
+) -> Result<i64, DbError> {
+    let client = pool.get().await?;
+    let row = client
+        .query_one(
+            "INSERT INTO users (username, email, password_hash, created_at) \
+             VALUES ($1, $2, $3, $4) RETURNING id",
+            &[&username, &email, &password_hash, &now_unix()],
+        )
+        .await?;
+    Ok(row.get(0))
 }
 
-pub async fn get_user_by_id(pool: &SqlitePool, id: i64) -> sqlx::Result<Option<User>> {
-    sqlx::query_as::<_, User>(
-        "SELECT id, username, email, password_hash, created_at FROM users WHERE id = ?",
-    )
-    .bind(id)
-    .fetch_optional(pool)
-    .await
+pub async fn get_user_by_id(pool: &Pool, id: i64) -> Result<Option<User>, DbError> {
+    let client = pool.get().await?;
+    let row = client
+        .query_opt(
+            "SELECT id, username, email, password_hash, created_at FROM users WHERE id = $1",
+            &[&id],
+        )
+        .await?;
+    Ok(row.map(|r| User {
+        id: r.get("id"),
+        username: r.get("username"),
+        email: r.get("email"),
+        password_hash: r.get("password_hash"),
+        created_at: r.get("created_at"),
+    }))
 }
 
-pub async fn get_user_by_username(pool: &SqlitePool, username: &str) -> sqlx::Result<Option<User>> {
-    sqlx::query_as::<_, User>(
-        "SELECT id, username, email, password_hash, created_at FROM users WHERE username = ?",
-    )
-    .bind(username)
-    .fetch_optional(pool)
-    .await
+pub async fn get_user_by_username(pool: &Pool, username: &str) -> Result<Option<User>, DbError> {
+    let client = pool.get().await?;
+    let row = client
+        .query_opt(
+            "SELECT id, username, email, password_hash, created_at FROM users WHERE username = $1",
+            &[&username],
+        )
+        .await?;
+    Ok(row.map(|r| User {
+        id: r.get("id"),
+        username: r.get("username"),
+        email: r.get("email"),
+        password_hash: r.get("password_hash"),
+        created_at: r.get("created_at"),
+    }))
 }
 
 // ── Game records ─────────────────────────────────────────────────────────────
 
 /// Creates a new game record when a room opens. Returns the record id.
 pub async fn insert_game_record(
-    pool: &SqlitePool,
+    pool: &Pool,
     game_id: &str,
     room_code: &str,
-) -> sqlx::Result<i64> {
-    let id = sqlx::query(
-        "INSERT INTO game_records (game_id, room_code, started_at) VALUES (?, ?, ?)",
-    )
-    .bind(game_id)
-    .bind(room_code)
-    .bind(now_unix())
-    .execute(pool)
-    .await?
-    .last_insert_rowid();
-    Ok(id)
+) -> Result<i64, DbError> {
+    let client = pool.get().await?;
+    let row = client
+        .query_one(
+            "INSERT INTO game_records (game_id, room_code, started_at) \
+             VALUES ($1, $2, $3) RETURNING id",
+            &[&game_id, &room_code, &now_unix()],
+        )
+        .await?;
+    Ok(row.get(0))
 }
 
 /// Stamps `ended_at` and stores the opaque result JSON supplied by the game.
 pub async fn close_game_record(
-    pool: &SqlitePool,
+    pool: &Pool,
     record_id: i64,
     result_json: Option<&str>,
-) -> sqlx::Result<()> {
+) -> Result<(), DbError> {
     // AND ended_at IS NULL prevents overwriting a result already set by POST /games/result
-    sqlx::query(
-        "UPDATE game_records SET ended_at = ?, result = ? WHERE id = ? AND ended_at IS NULL",
-    )
-    .bind(now_unix())
-    .bind(result_json)
-    .bind(record_id)
-    .execute(pool)
-    .await?;
+    let client = pool.get().await?;
+    client
+        .execute(
+            "UPDATE game_records SET ended_at = $1, result = $2 \
+             WHERE id = $3 AND ended_at IS NULL",
+            &[&now_unix(), &result_json, &record_id],
+        )
+        .await?;
     Ok(())
 }
 
 /// Records a player's participation in a game. `user_id` is `None` for anonymous players.
 pub async fn insert_participant(
-    pool: &SqlitePool,
+    pool: &Pool,
     record_id: i64,
     user_id: Option<i64>,
     player_id: u16,
     outcome: Option<&str>,
-) -> sqlx::Result<()> {
-    sqlx::query(
-        "INSERT OR IGNORE INTO game_participants (game_record_id, user_id, player_id, outcome)
-         VALUES (?, ?, ?, ?)",
-    )
-    .bind(record_id)
-    .bind(user_id)
-    .bind(player_id as i64)
-    .bind(outcome)
-    .execute(pool)
-    .await?;
+) -> Result<(), DbError> {
+    let client = pool.get().await?;
+    client
+        .execute(
+            "INSERT INTO game_participants (game_record_id, user_id, player_id, outcome) \
+             VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
+            &[&record_id, &user_id, &(player_id as i64), &outcome],
+        )
+        .await?;
     Ok(())
 }
 
 /// Returns win/loss/draw counts for a user. All values are 0 when the user has no games.
-pub async fn get_user_stats(pool: &SqlitePool, user_id: i64) -> sqlx::Result<UserStats> {
-    sqlx::query_as::<_, UserStats>(
-        "SELECT
-             COUNT(*) as total,
-             COALESCE(SUM(CASE WHEN outcome = 'win'  THEN 1 ELSE 0 END), 0) as wins,
-             COALESCE(SUM(CASE WHEN outcome = 'loss' THEN 1 ELSE 0 END), 0) as losses,
-             COALESCE(SUM(CASE WHEN outcome = 'draw' THEN 1 ELSE 0 END), 0) as draws
-         FROM game_participants
-         WHERE user_id = ?",
-    )
-    .bind(user_id)
-    .fetch_one(pool)
-    .await
+pub async fn get_user_stats(pool: &Pool, user_id: i64) -> Result<UserStats, DbError> {
+    let client = pool.get().await?;
+    let row = client
+        .query_one(
+            "SELECT
+                 COUNT(*) as total,
+                 COALESCE(SUM(CASE WHEN outcome = 'win'  THEN 1 ELSE 0 END), 0::BIGINT) as wins,
+                 COALESCE(SUM(CASE WHEN outcome = 'loss' THEN 1 ELSE 0 END), 0::BIGINT) as losses,
+                 COALESCE(SUM(CASE WHEN outcome = 'draw' THEN 1 ELSE 0 END), 0::BIGINT) as draws
+             FROM game_participants
+             WHERE user_id = $1",
+            &[&user_id],
+        )
+        .await?;
+    Ok(UserStats {
+        total: row.get("total"),
+        wins: row.get("wins"),
+        losses: row.get("losses"),
+        draws: row.get("draws"),
+    })
 }
 
 /// Returns a paginated list of games a user participated in, newest first.
 pub async fn get_user_games(
-    pool: &SqlitePool,
+    pool: &Pool,
     user_id: i64,
     page: i64,
     per_page: i64,
-) -> sqlx::Result<Vec<GameSummary>> {
-    sqlx::query_as::<_, GameSummary>(
-        "SELECT gr.id, gr.game_id, gr.room_code, gr.started_at, gr.ended_at, gr.result, gp.outcome
-         FROM game_records gr
-         JOIN game_participants gp ON gp.game_record_id = gr.id
-         WHERE gp.user_id = ?
-         ORDER BY gr.started_at DESC
-         LIMIT ? OFFSET ?",
-    )
-    .bind(user_id)
-    .bind(per_page)
-    .bind(page * per_page)
-    .fetch_all(pool)
-    .await
+) -> Result<Vec<GameSummary>, DbError> {
+    let client = pool.get().await?;
+    let rows = client
+        .query(
+            "SELECT gr.id, gr.game_id, gr.room_code, gr.started_at, gr.ended_at, gr.result, gp.outcome
+             FROM game_records gr
+             JOIN game_participants gp ON gp.game_record_id = gr.id
+             WHERE gp.user_id = $1
+             ORDER BY gr.started_at DESC
+             LIMIT $2 OFFSET $3",
+            &[&user_id, &per_page, &(page * per_page)],
+        )
+        .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| GameSummary {
+            id: r.get("id"),
+            game_id: r.get("game_id"),
+            room_code: r.get("room_code"),
+            started_at: r.get("started_at"),
+            ended_at: r.get("ended_at"),
+            result: r.get("result"),
+            outcome: r.get("outcome"),
+        })
+        .collect())
 }