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

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

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
);
"#;