//! 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 { 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> { 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> { 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> { 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::("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 { 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> { 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::("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> { 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> { 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> { 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> { 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> { 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::("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> { 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> { 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::("json")).ok())) } // ---- player-vendor market index (Protocol 3.0) ---- /// Upserts one vendor's whole listing, keyed by serial. Fed from `vendor.listing`, which the /// shard emits as an authoritative per-vendor frame — so this replaces the row outright rather /// than merging anything. /// /// The items ride inside `json` and are deliberately NOT normalized into a `vendor_items` /// table. The sidecar's job for the market is outage resilience (`PROTOCOL_2.md` §12.2) — hand /// the website back what the shard last said — not search. Search lives in MariaDB on the /// website side, where the query surface, the indexes and the cliloc-resolved display names /// already are; a second search implementation here would be one more thing to keep in step /// with it for no reader. #[allow(clippy::too_many_arguments)] pub async fn upsert_vendor( &self, serial: &str, shop_name: Option<&str>, owner_name: Option<&str>, map: Option<&str>, x: Option, y: Option, region: Option<&str>, count: Option, json: &str, t: i64, ) -> anyhow::Result<()> { sqlx::query( "INSERT INTO vendors (serial, shop_name, owner_name, map, x, y, region, count, json, updated_t) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(serial) DO UPDATE SET shop_name = excluded.shop_name, owner_name = excluded.owner_name, map = excluded.map, x = excluded.x, y = excluded.y, region = excluded.region, count = excluded.count, json = excluded.json, updated_t = excluded.updated_t", ) .bind(serial) .bind(shop_name) .bind(owner_name) .bind(map) .bind(x) .bind(y) .bind(region) .bind(count) .bind(json) .bind(t) .execute(&self.pool) .await?; Ok(()) } /// Drops one vendor from the index. Fed from `vendor.listing.remove` — a vendor dismissed, /// expired, or whose owner switched off its in-game Vendor Search flag. pub async fn delete_vendor(&self, serial: &str) -> anyhow::Result<()> { sqlx::query("DELETE FROM vendors WHERE serial = ?") .bind(serial) .execute(&self.pool) .await?; Ok(()) } /// One page of the index, ordered by serial. /// /// Paged where the other boards are not, and the ordering is why it can be: a whole-world /// market is the one board that does not fit in a response. Ordering by SERIAL rather than by /// shop name is deliberate — the page is a snapshot cursor for the website's reconnect /// backfill, and a serial is stable while a shop name is renameable, so a rename mid-backfill /// cannot make a vendor skip or repeat a page. pub async fn vendors_page(&self, limit: i64, offset: i64) -> anyhow::Result> { let limit = limit.clamp(1, 1000); let offset = offset.max(0); let rows = sqlx::query("SELECT json FROM vendors ORDER BY serial LIMIT ? OFFSET ?") .bind(limit) .bind(offset) .fetch_all(&self.pool) .await?; Ok(parse_json_column(rows)) } /// How many vendors the index holds, so a paging caller knows when to stop. pub async fn vendors_count(&self) -> anyhow::Result { let row = sqlx::query("SELECT COUNT(*) AS n FROM vendors") .fetch_one(&self.pool) .await?; Ok(row.get::("n")) } // ---- 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> { 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) -> Vec { rows.into_iter() .filter_map(|r| serde_json::from_str(&r.get::("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 ); -- Player-vendor market index (Protocol 3.0). One row per vendor, holding the whole authoritative -- `vendor.listing` frame including its items. The hoisted columns exist for the ORDER BY and for -- an operator eyeballing the table; nothing here is searched, because search is the website's job -- (see upsert_vendor). Rows are dropped on `vendor.listing.remove`. CREATE TABLE IF NOT EXISTS vendors ( serial TEXT PRIMARY KEY, shop_name TEXT, owner_name TEXT, map TEXT, x INTEGER, y INTEGER, region TEXT, count INTEGER, 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 ); "#;