link/sidecar/src/store.rs

//! SQLite persistence: event history, the economy series, cached profiles, and the account-link
//! map. This is what lets the website read the past without asking the shard, and what survives a
//! sidecar restart.
//!
//! The event loop writes every live event here as it broadcasts it; REST read endpoints query here
//! instead of round-tripping the shard. Links and profiles are written from the REST reply paths
//! (`link.ok`, `char.profile`), which are RPC replies and never hit the broadcast stream.

use std::str::FromStr;

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

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

impl Store {
    /// Opens (creating if absent) the SQLite database and ensures the schema exists.
    pub async fn open(path: &str) -> anyhow::Result<Self> {
        let opts =
            SqliteConnectOptions::from_str(&format!("sqlite://{path}"))?.create_if_missing(true);

        let pool = SqlitePoolOptions::new()
            .max_connections(4)
            .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))
    }

    /// The money-supply series: the `economy.supply` events, newest first.
    pub async fn economy(&self, limit: i64) -> anyhow::Result<Vec<Value>> {
        self.recent(Some("economy.supply"), limit).await
    }

    pub async fn record_link(
        &self,
        account: &str,
        website_user_id: &str,
        t: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO links (account, website_user_id, linked_t) VALUES (?, ?, ?)
             ON CONFLICT(account) DO UPDATE SET website_user_id = excluded.website_user_id, linked_t = excluded.linked_t",
        )
        .bind(account)
        .bind(website_user_id)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    pub async fn get_link(&self, account: &str) -> anyhow::Result<Option<String>> {
        let row = sqlx::query("SELECT website_user_id FROM links WHERE account = ?")
            .bind(account)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.map(|r| r.get::<String, _>("website_user_id")))
    }

    /// Drops the mirrored link row so event attribution stops immediately, without waiting on the
    /// shard. Returns the number of rows removed (0 if the account was not linked here).
    pub async fn record_unlink(&self, account: &str) -> anyhow::Result<u64> {
        let res = sqlx::query("DELETE FROM links WHERE account = ?")
            .bind(account)
            .execute(&self.pool)
            .await?;
        Ok(res.rows_affected())
    }

    pub async fn cache_profile(
        &self,
        serial: &str,
        account: Option<&str>,
        name: Option<&str>,
        json: &str,
        t: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO profiles (serial, account, name, json, updated_t) VALUES (?, ?, ?, ?, ?)
             ON CONFLICT(serial) DO UPDATE SET account = excluded.account, name = excluded.name, json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(serial)
        .bind(account)
        .bind(name)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    pub async fn get_cached_profile(&self, serial: &str) -> anyhow::Result<Option<Value>> {
        let row = sqlx::query("SELECT json FROM profiles WHERE serial = ?")
            .bind(serial)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
    }

    /// Upserts one champion-spawn's latest state, keyed by serial. Fed from the `champ.update`
    /// stream; this table is the live board the website reads, so there is exactly one row per
    /// spawn and it always holds the most recent snapshot.
    pub async fn upsert_champ(
        &self,
        serial: &str,
        status: Option<&str>,
        name: Option<&str>,
        json: &str,
        t: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO champs (serial, status, name, json, updated_t) VALUES (?, ?, ?, ?, ?)
             ON CONFLICT(serial) DO UPDATE SET status = excluded.status, name = excluded.name, json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(serial)
        .bind(status)
        .bind(name)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Drops one spawn from the board. Fed from the `champ.remove` stream: a controller that was
    /// deleted, or a transient sea boss that was slain, leaves the board this way.
    pub async fn delete_champ(&self, serial: &str) -> anyhow::Result<()> {
        sqlx::query("DELETE FROM champs WHERE serial = ?")
            .bind(serial)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// The full champion-spawn board: every spawn's latest snapshot. Ordered by name so the site
    /// gets a stable list.
    pub async fn champs_all(&self) -> anyhow::Result<Vec<Value>> {
        let rows = sqlx::query("SELECT json FROM champs ORDER BY name, serial")
            .fetch_all(&self.pool)
            .await?;
        Ok(parse_json_column(rows))
    }

    // ---- guild board (Protocol 2.0) ----

    /// Upserts one guild's latest state, keyed by guild id. Fed from `guild.update`; one row per
    /// guild, always the most recent snapshot. This is the board the website reads on load.
    pub async fn upsert_guild(
        &self,
        id: i64,
        name: Option<&str>,
        json: &str,
        t: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO guilds (id, name, json, updated_t) VALUES (?, ?, ?, ?)
             ON CONFLICT(id) DO UPDATE SET name = excluded.name, json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(id)
        .bind(name)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Drops one guild from the board. Fed from `guild.remove` (a disband or a removed guild).
    pub async fn delete_guild(&self, id: i64) -> anyhow::Result<()> {
        sqlx::query("DELETE FROM guilds WHERE id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// The full guild board: every guild's latest snapshot, ordered by name.
    pub async fn guilds_all(&self) -> anyhow::Result<Vec<Value>> {
        let rows = sqlx::query("SELECT json FROM guilds ORDER BY name, id")
            .fetch_all(&self.pool)
            .await?;
        Ok(parse_json_column(rows))
    }

    // ---- governor board (Protocol 2.0) ----

    /// Upserts one city's latest governance state, keyed by city name. Fed from `city.update`.
    pub async fn upsert_governor(&self, city: &str, json: &str, t: i64) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO governors (city, json, updated_t) VALUES (?, ?, ?)
             ON CONFLICT(city) DO UPDATE SET json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(city)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// The full governor board: every city's latest governance snapshot, ordered by city.
    pub async fn governors_all(&self) -> anyhow::Result<Vec<Value>> {
        let rows = sqlx::query("SELECT json FROM governors ORDER BY city")
            .fetch_all(&self.pool)
            .await?;
        Ok(parse_json_column(rows))
    }

    // ---- house registry (Protocol 2.0) ----

    /// Upserts one house's latest state, keyed by serial. Fed from `house.update`.
    pub async fn upsert_house(
        &self,
        serial: &str,
        name: Option<&str>,
        json: &str,
        t: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO houses (serial, name, json, updated_t) VALUES (?, ?, ?, ?)
             ON CONFLICT(serial) DO UPDATE SET name = excluded.name, json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(serial)
        .bind(name)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Drops one house from the registry. Fed from `house.remove` (demolished / traded away).
    pub async fn delete_house(&self, serial: &str) -> anyhow::Result<()> {
        sqlx::query("DELETE FROM houses WHERE serial = ?")
            .bind(serial)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// The full house registry: every house's latest snapshot, ordered by name then serial.
    pub async fn houses_all(&self) -> anyhow::Result<Vec<Value>> {
        let rows = sqlx::query("SELECT json FROM houses ORDER BY name, serial")
            .fetch_all(&self.pool)
            .await?;
        Ok(parse_json_column(rows))
    }

    // ---- shard ruleset (Protocol 3.0) ----

    /// Stores the shard's published ruleset. A singleton (`id = 1`): the shard emits one
    /// `world.ruleset` frame per connect describing how it is configured, and only the latest one
    /// matters. `rev` is the shard's FNV-1a of the body, kept so a reader can tell "same ruleset,
    /// re-sent on reconnect" from "the operator changed something" without diffing the JSON.
    pub async fn upsert_ruleset(&self, rev: Option<&str>, json: &str, t: i64) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO ruleset (id, rev, json, updated_t) VALUES (1, ?, ?, ?)
             ON CONFLICT(id) DO UPDATE SET rev = excluded.rev, json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(rev)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// The stored ruleset, or `None` if the shard has never published one. Returning `None` rather
    /// than an empty object is deliberate: "not published yet" and "published, everything off" are
    /// different answers and the website renders them differently.
    pub async fn ruleset(&self) -> anyhow::Result<Option<Value>> {
        let row = sqlx::query("SELECT json FROM ruleset WHERE id = 1")
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
    }

    // ---- points / loyalty boards (Protocol 3.0) ----

    /// Upserts one system's leaderboard, keyed by its `PointsType` name (`QueensLoyalty`,
    /// `CleanUpBritannia`, …). Fed from `points.board`; one row per system, always the most recent
    /// top-N snapshot.
    ///
    /// There is no matching delete, and that is deliberate rather than an omission: the shard's set
    /// of point systems is fixed at startup by `PointsSystem.Configure`, so a system cannot vanish
    /// at runtime and the plugin emits no `points.remove`. Same argument the governor board makes.
    pub async fn upsert_points_board(
        &self,
        system: &str,
        name: Option<&str>,
        json: &str,
        t: i64,
    ) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO points_boards (system, name, json, updated_t) VALUES (?, ?, ?, ?)
             ON CONFLICT(system) DO UPDATE SET name = excluded.name, json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(system)
        .bind(name)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Every system's latest board, ordered by display name then system key. Systems the shard has
    /// never published are simply absent — the website renders the set it is given.
    pub async fn points_boards_all(&self) -> anyhow::Result<Vec<Value>> {
        let rows = sqlx::query("SELECT json FROM points_boards ORDER BY name, system")
            .fetch_all(&self.pool)
            .await?;
        Ok(parse_json_column(rows))
    }

    /// One system's board, or `None` when that system has never published one. `None` is a real
    /// answer (an unknown system name, or one the operator excluded via `Bridge.cfg PointsSystems`),
    /// which the website turns into a 404 rather than an empty board.
    pub async fn points_board(&self, system: &str) -> anyhow::Result<Option<Value>> {
        let row = sqlx::query("SELECT json FROM points_boards WHERE system = ?")
            .bind(system)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
    }

    // ---- Town Cryer news (Protocol 2.1) ----

    /// Stores/replaces one external news article (the `news.add` command json), keyed by id. The
    /// website is the source of truth; this lets the sidecar replay the set to the shard on reconnect
    /// (the shard does not persist NewsEntries across a reboot).
    pub async fn upsert_news(&self, id: &str, json: &str, t: i64) -> anyhow::Result<()> {
        sqlx::query(
            "INSERT INTO news (id, json, updated_t) VALUES (?, ?, ?)
             ON CONFLICT(id) DO UPDATE SET json = excluded.json, updated_t = excluded.updated_t",
        )
        .bind(id)
        .bind(json)
        .bind(t)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Removes one external news article.
    pub async fn delete_news(&self, id: &str) -> anyhow::Result<()> {
        sqlx::query("DELETE FROM news WHERE id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// Every stored external news article (as its `news.add` command), oldest first so a replay
    /// re-inserts them in the same order the website added them.
    pub async fn news_all(&self) -> anyhow::Result<Vec<Value>> {
        let rows = sqlx::query("SELECT json FROM news ORDER BY updated_t")
            .fetch_all(&self.pool)
            .await?;
        Ok(parse_json_column(rows))
    }
}

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()
}

const SCHEMA: &str = r#"
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 ON events(kind, id);

CREATE TABLE IF NOT EXISTS links (
    account         TEXT PRIMARY KEY,
    website_user_id TEXT NOT NULL,
    linked_t        INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS profiles (
    serial    TEXT PRIMARY KEY,
    account   TEXT,
    name      TEXT,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS champs (
    serial    TEXT PRIMARY KEY,
    status    TEXT,
    name      TEXT,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS guilds (
    id        INTEGER PRIMARY KEY,
    name      TEXT,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS governors (
    city      TEXT PRIMARY KEY,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS houses (
    serial    TEXT PRIMARY KEY,
    name      TEXT,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS news (
    id        TEXT PRIMARY KEY,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

-- Points/loyalty leaderboards (Protocol 3.0). One row per point system, keyed by the shard's
-- own PointsType name; `name` is the resolved display name, hoisted only for the ORDER BY.
CREATE TABLE IF NOT EXISTS points_boards (
    system    TEXT PRIMARY KEY,
    name      TEXT,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);

-- The shard's published ruleset (Protocol 3.0). Singleton: the CHECK is what makes it one,
-- so an upsert can target id = 1 unconditionally and no second row can ever appear.
CREATE TABLE IF NOT EXISTS ruleset (
    id        INTEGER PRIMARY KEY CHECK (id = 1),
    rev       TEXT,
    json      TEXT NOT NULL,
    updated_t INTEGER NOT NULL
);
"#;