Rust-Link/sidecar/src/store.rs

//! SQLite persistence: the event history, and the last thing the game said about itself.
//!
//! This is what lets the website read the past without asking the game, and what survives a sidecar
//! restart. The event loop writes every live event here as it broadcasts it; REST reads query here
//! instead of round-tripping the plugin.
//!
//! **The sidecar defines no schema for a frame's contents.** Events are persisted whole, as the
//! JSON text that arrived, with only `t` and `kind` lifted out for indexing. That is the
//! dumb-forwarder property doing real work: a protocol version that adds fields to an event needs
//! no change here, and only a version that adds a *new indexed column* ever needs a migration.

use std::path::Path;

use serde_json::Value;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{Row, SqlitePool};
use tracing::info;

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS events (
    id    INTEGER PRIMARY KEY AUTOINCREMENT,
    t     INTEGER NOT NULL,
    kind  TEXT    NOT NULL,
    json  TEXT    NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_kind_id ON events (kind, id DESC);
CREATE INDEX IF NOT EXISTS idx_events_t ON events (t);

-- Exactly one row, id 1: the most recent server.hello. A board, in the sense chapter 4 uses the
-- word — current state with one producer, re-sent on every connect — rather than a history.
CREATE TABLE IF NOT EXISTS server_state (
    id      INTEGER PRIMARY KEY CHECK (id = 1),
    t       INTEGER NOT NULL,
    json    TEXT    NOT NULL
);
";

#[derive(Clone)]
pub struct Store {
    pool: SqlitePool,
}

impl Store {
    /// Opens (creating if absent) the SQLite database and ensures the schema exists.
    ///
    /// `path` is a filesystem path, handed to sqlx as one. It is deliberately **not** formatted
    /// into a `sqlite://` URL first: that spelling is parsed as a URL, so it percent-decodes the
    /// path and splits it on `?`. Under an installed layout the path is absolute and chosen by the
    /// operator — `C:\ProgramData\RunicGateway\rust-link.db`, or something under a home directory
    /// with a `%` or `#` in it — and a URL round-trip silently opens a *different* file.
    pub async fn open(path: &str) -> anyhow::Result<Self> {
        // A service unit can name a data directory that does not exist yet; creating it here means
        // one less way for a fresh install to fail on first start.
        if let Some(dir) = Path::new(path).parent() {
            if !dir.as_os_str().is_empty() && !dir.exists() {
                std::fs::create_dir_all(dir)?;
            }
        }

        let opts = SqliteConnectOptions::new()
            .filename(path)
            .create_if_missing(true);

        // An in-memory database is **per connection**, not per process: every connection the pool
        // opens gets its own empty one, so a second pooled connection finds none of the schema the
        // first created. It presents as `no such table` from a random subset of queries, which is
        // as confusing a failure as this file has. A single connection is the only coherent
        // reading of `:memory:`, and it is what makes it usable at all.
        let max = if is_in_memory(path) { 1 } else { 4 };

        let pool = SqlitePoolOptions::new()
            .max_connections(max)
            .connect_with(opts)
            .await?;

        sqlx::query(SCHEMA).execute(&pool).await?;
        info!(%path, "store ready");
        Ok(Self { pool })
    }

    /// Cheap liveness check for the health endpoint.
    pub async fn ping(&self) -> anyhow::Result<()> {
        sqlx::query("SELECT 1").execute(&self.pool).await?;
        Ok(())
    }

    /// Appends one live event. Failures are logged by the caller; persistence must never block the
    /// live feed.
    pub async fn insert_event(&self, t: i64, kind: &str, json: &str) -> anyhow::Result<()> {
        sqlx::query("INSERT INTO events (t, kind, json) VALUES (?, ?, ?)")
            .bind(t)
            .bind(kind)
            .bind(json)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// Most-recent events, newest first, optionally filtered by kind.
    pub async fn recent(&self, kind: Option<&str>, limit: i64) -> anyhow::Result<Vec<Value>> {
        let limit = limit.clamp(1, 1000);
        let rows = match kind {
            Some(k) => {
                sqlx::query("SELECT json FROM events WHERE kind = ? ORDER BY id DESC LIMIT ?")
                    .bind(k)
                    .bind(limit)
                    .fetch_all(&self.pool)
                    .await?
            }
            None => {
                sqlx::query("SELECT json FROM events ORDER BY id DESC LIMIT ?")
                    .bind(limit)
                    .fetch_all(&self.pool)
                    .await?
            }
        };
        Ok(parse_json_column(rows))
    }

    /// Replaces the one `server_state` row. Called for every `server.hello`, which the plugin sends
    /// on every connect — so this is an upsert by construction, not by accident.
    pub async fn put_server_state(&self, t: i64, json: &str) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO server_state (id, t, json) VALUES (1, ?, ?)
             ON CONFLICT(id) DO UPDATE SET t = excluded.t, json = excluded.json",
        )
        .bind(t)
        .bind(json)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// The last thing the game said about itself, or `None` if it has never connected.
    ///
    /// This is the read that makes the website render while the game is off, which is the whole
    /// reason the sidecar holds a database at all.
    pub async fn server_state(&self) -> anyhow::Result<Option<Value>> {
        let row = sqlx::query("SELECT json FROM server_state WHERE id = 1")
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
    }
}

/// Whether this path names an in-memory database rather than a file. Covers the bare `:memory:`
/// spelling and the `file:` URI form that carries `mode=memory`.
fn is_in_memory(path: &str) -> bool {
    path == ":memory:" || (path.starts_with("file:") && path.contains("mode=memory"))
}

fn parse_json_column(rows: Vec<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
    rows.into_iter()
        .filter_map(|r| serde_json::from_str(&r.get::<String, _>("json")).ok())
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    async fn store() -> Store {
        Store::open(":memory:").await.unwrap()
    }

    /// The trap this file's pool sizing exists for: a multi-connection pool over `:memory:` hands
    /// out empty databases. Asserting the *pool* is what makes the reason visible; asserting only
    /// that a query works would pass again the moment someone "tidied" the sizing back.
    #[tokio::test]
    async fn an_in_memory_store_uses_exactly_one_connection() {
        assert!(is_in_memory(":memory:"));
        assert!(is_in_memory("file:x?mode=memory&cache=shared"));
        assert!(!is_in_memory("rust-link.db"));
        assert!(!is_in_memory("file:/var/lib/rg/rust-link.db"));

        let s = store().await;
        assert_eq!(s.pool.options().get_max_connections(), 1);
    }

    /// `sqlx::query` over a multi-statement string is the kind of thing that quietly runs only the
    /// first statement. Both tables and both reads have to work on a real file, under the pool
    /// size production uses.
    #[tokio::test]
    async fn the_whole_schema_is_created_on_a_pooled_file_store() {
        let dir = std::env::temp_dir().join(format!("rust-link-test-{}", std::process::id()));
        let path = dir.join("schema.db");
        let _ = std::fs::remove_dir_all(&dir);

        let s = Store::open(path.to_str().unwrap()).await.unwrap();
        assert_eq!(s.pool.options().get_max_connections(), 4);

        s.insert_event(1, "k", "{}").await.unwrap();
        s.put_server_state(1, "{}").await.unwrap();
        assert_eq!(s.recent(None, 10).await.unwrap().len(), 1);
        assert!(s.server_state().await.unwrap().is_some());

        drop(s);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn events_come_back_newest_first_and_filter_by_kind() {
        let s = store().await;
        s.insert_event(1, "server.hello", &json!({"n": 1}).to_string())
            .await
            .unwrap();
        s.insert_event(2, "other", &json!({"n": 2}).to_string())
            .await
            .unwrap();
        s.insert_event(3, "server.hello", &json!({"n": 3}).to_string())
            .await
            .unwrap();

        let all = s.recent(None, 10).await.unwrap();
        assert_eq!(all.len(), 3);
        assert_eq!(all[0]["n"], 3);

        let hellos = s.recent(Some("server.hello"), 10).await.unwrap();
        assert_eq!(hellos.len(), 2);
        assert_eq!(hellos[0]["n"], 3);
    }

    /// An absent board reads as `None`, not as an empty object. A caller must be able to tell
    /// "the game has never connected" from "the game connected and said nothing" — collapsing the
    /// two is how a site ends up rendering a server that does not exist.
    #[tokio::test]
    async fn server_state_is_absent_until_a_hello_arrives() {
        let s = store().await;
        assert!(s.server_state().await.unwrap().is_none());

        s.put_server_state(1, &json!({"serverId": "main", "players": 0}).to_string())
            .await
            .unwrap();
        assert_eq!(s.server_state().await.unwrap().unwrap()["serverId"], "main");
    }

    /// The board holds exactly one row however many times the plugin reconnects.
    #[tokio::test]
    async fn a_second_hello_replaces_the_first() {
        let s = store().await;
        s.put_server_state(1, &json!({"bootId": "a"}).to_string())
            .await
            .unwrap();
        s.put_server_state(2, &json!({"bootId": "b"}).to_string())
            .await
            .unwrap();

        assert_eq!(s.server_state().await.unwrap().unwrap()["bootId"], "b");
        let count: i64 = sqlx::query("SELECT COUNT(*) AS c FROM server_state")
            .fetch_one(&s.pool)
            .await
            .unwrap()
            .get("c");
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn the_limit_is_clamped_rather_than_trusted() {
        let s = store().await;
        for i in 0..5 {
            s.insert_event(i, "k", &json!({"i": i}).to_string())
                .await
                .unwrap();
        }
        // 0 and negatives would otherwise mean "no rows" and "SQLite's unlimited" respectively.
        assert_eq!(s.recent(None, 0).await.unwrap().len(), 1);
        assert_eq!(s.recent(None, -1).await.unwrap().len(), 1);
        assert_eq!(s.recent(None, 100_000).await.unwrap().len(), 5);
    }
}