Sidecar: SQLite persistence

store.rs (sqlx/sqlite) makes the data durable and queryable over time. Three
tables: events (the full live stream, append-only), links (account <-> website
user, mirrored from link.ok), profiles (last-known character sheet, cached from
char.profile). The event loop persists every live event before broadcasting it;
pong is dropped as ephemeral chatter.

New read endpoints served from the DB rather than the shard: GET /history
(optionally ?kind=), GET /economy (the money-supply series), GET /link/{account}.
GET /char/serial/{serial} now falls back to the cached profile when the shard is
unreachable, so an already-viewed character still renders during an outage;
link.confirm mirrors a successful link into the store.

Verified end to end: 10 economy.supply snapshots and the rest of the live stream
persisted and served via /history and /economy; the data survived a sidecar
restart (14 events still present, and the shard reconnected to the new sidecar);
and with the shard killed, a cached profile returned at HTTP 200 while an uncached
query failed cleanly at 503.

The sidecar is feature-complete: shard link, WebSocket feed, REST queries, and
persistence all work end-to-end against the live shard.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

sidecar/.gitignore

@@ -1 +1,3 @@
 /target
+*.db
+*.db-*

sidecar/Cargo.toml

@@ -12,6 +12,7 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 anyhow = "1"
 axum = { version = "0.7", features = ["ws"] }
+sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite"] }
 
 [profile.release]
 opt-level = 2

sidecar/README.md

@@ -25,7 +25,9 @@ Binds `127.0.0.1:7788` and waits for the shard to connect. Boot the shard (or it
 | Shard link (`shard.rs`) | **done** — accepts the shard, reads events, sends commands, re-accepts on disconnect. Verified against the live shard: received `server.hello`, round-tripped a `ping`→`pong`, and reconnected after a sidecar restart. |
 | WebSocket feed (`web.rs`) | **done** — `/ws` fans every shard event out to connected clients via a `broadcast`. Verified: a WS client received `ws.hello` then live `pong` events relayed from the shard. Live-only, no replay. |
 | REST queries (`rpc.rs` + `web.rs`) | **done** — synchronous queries and commands, correlated to shard replies by id. Verified end-to-end against the live shard, success and error paths. |
-| SQLite persistence (event history, economy, cached profiles, link map) | not started |
+| SQLite persistence (`store.rs`) | **done** — every live event persisted; history/economy served from the DB; profiles cached with shard-down fallback; link map. Verified: data survived a sidecar restart, and a cached profile served at 200 with the shard killed. |
+
+**The sidecar is feature-complete.** All four pieces work end-to-end against the live shard.
 
 The web server binds `127.0.0.1:8080` by default (`WEB_ADDR` in `main.rs`). Widen the bind and add auth before exposing it off-host.
 
@@ -42,15 +44,19 @@ The web server binds `127.0.0.1:8080` by default (`WEB_ADDR` in `main.rs`). Wide
 | POST | `/link/confirm` `{code, websiteUserId}` | `link.confirm` | `link.ok` / `link.error` |
 | POST | `/towncrier` `{id, lines, durationSec}` | `towncrier.add` | `towncrier.ok` / `towncrier.error` |
 | DELETE | `/towncrier/{id}` | `towncrier.remove` | `towncrier.ok` / `towncrier.error` |
+| GET | `/link/{account}` | — (reads store) | `{account, websiteUserId}` or 404 |
+| GET | `/history?kind=&limit=` | — (reads store) | `{events: [...]}` newest first |
+| GET | `/economy?limit=` | — (reads store) | `{series: [...]}` supply snapshots |
 
-A shard `*.error` reply maps to HTTP 404 (unknown/not-found) or 400 (bad request). No shard connected → 503; no reply within 10 s → 504.
+A shard `*.error` reply maps to HTTP 404 (unknown/not-found) or 400 (bad request). No shard connected → 503; no reply within 10 s → 504. `GET /char/serial/{serial}` falls back to the cached profile when the shard is unreachable, so an already-viewed character still renders during an outage.
 
 ## Design
 
 - **`shard.rs`** — `serve()` binds the listener and accepts shard connections in a loop. Each connection splits into read/write halves: the read half parses newline-JSON into `ShardEvent { kind, value }` and forwards them; the write half drains an mpsc of command lines. `ShardHandle::send` posts a command to whichever shard is currently connected, and **drops with a warning if none is** — a website query during a shard outage should fail fast and retry, not queue behind a reconnect. Live *events* that must survive an outage are buffered by the shard, not here.
 - **`web.rs`** — the website-facing HTTP surface (axum). `AppState` holds the `broadcast::Sender<String>`; each `/ws` client subscribes and forwards every event as a text frame. A client that lags past the broadcast buffer is warned and kept live (it just misses events) rather than stalling the others. This side *may* be exposed beyond loopback — it is the gatekeeper, so add auth when you do.
 - **`rpc.rs`** — request/reply correlation over the one shard socket. A REST call registers a pending entry under a correlation id, sends the command, and awaits the reply (10 s timeout). The event loop routes any incoming line whose id is pending back to the waiter; everything else flows on as a live event. Recognizes three correlation fields, matching what the plugin echoes: `reqId` (queries), `code` (link), `id` (town-crier).
-- **`main.rs`** — wires it together: the shard event loop first tries to route each line as an RPC reply; if it isn't one, the line is a live event, logged and broadcast to WS. Later phases persist to SQLite here.
+- **`store.rs`** — SQLite (`sqlx`). Three tables: `events` (the full live stream, append-only), `links` (account ↔ website user, mirrored from `link.ok`), `profiles` (last-known character sheet, cached from `char.profile`). History and economy read here instead of the shard; `pong` is dropped as ephemeral chatter. DB file defaults to `uo-link.db` (`DB_PATH` in `main.rs`), gitignored.
+- **`main.rs`** — wires it together: the shard event loop first tries to route each line as an RPC reply; if it isn't one, the line is a live event — logged, persisted, and broadcast to WS.
 
 ## Wire protocol
 

sidecar/src/main.rs

@@ -5,6 +5,7 @@
 
 mod rpc;
 mod shard;
+mod store;
 mod web;
 
 use tokio::sync::{broadcast, mpsc};
@@ -13,6 +14,7 @@ use tracing_subscriber::EnvFilter;
 
 const SHARD_ADDR: &str = "127.0.0.1:7788";
 const WEB_ADDR: &str = "127.0.0.1:8080";
+const DB_PATH: &str = "uo-link.db";
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -29,11 +31,15 @@ async fn main() -> anyhow::Result<()> {
     // Request/reply correlation for REST queries.
     let rpc = rpc::Rpc::new();
 
+    // Durable store: event history, economy series, cached profiles, link map.
+    let store = store::Store::open(DB_PATH).await?;
+
     // Website-facing HTTP server.
     let web_state = web::AppState {
         events: bcast_tx.clone(),
         shard: handle.clone(),
         rpc: rpc.clone(),
+        store: store.clone(),
     };
     tokio::spawn(async move {
         if let Err(e) = web::serve(WEB_ADDR, web_state).await {
@@ -42,9 +48,10 @@ async fn main() -> anyhow::Result<()> {
     });
 
     // Event loop: a line that correlates to a pending REST call is a reply — route it to the
-    // waiting caller and stop. Everything else is a live event: log it and broadcast it.
+    // waiting caller and stop. Everything else is a live event: log it, persist it, broadcast it.
     let feed_tx = bcast_tx.clone();
     let route_rpc = rpc.clone();
+    let event_store = store.clone();
     let mut total: u64 = 0;
     tokio::spawn(async move {
         while let Some(ev) = event_rx.recv().await {
@@ -61,6 +68,15 @@ async fn main() -> anyhow::Result<()> {
                 _ => tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value),
             }
 
+            // Persist, then broadcast. `pong` and `ws.hello` are ephemeral chatter, not history.
+            if ev.kind != "pong" {
+                let t = ev.value.get("t").and_then(|v| v.as_i64()).unwrap_or_else(now_ms);
+                let text = ev.value.to_string();
+                if let Err(e) = event_store.insert_event(t, &ev.kind, &text).await {
+                    tracing::warn!(error = %e, "failed to persist event");
+                }
+            }
+
             let _ = feed_tx.send(ev.value.to_string());
         }
     });
@@ -83,6 +99,14 @@ async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
+fn now_ms() -> i64 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
 fn init_tracing() {
     tracing_subscriber::fmt()
         .with_env_filter(

sidecar/src/store.rs

@@ -0,0 +1,155 @@
+//! 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 })
+    }
+
+    /// 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()))
+    }
+}
+
+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
+);
+"#;

sidecar/src/web.rs

@@ -8,18 +8,20 @@ use std::time::Duration;
 
 use axum::{
     extract::ws::{Message, WebSocket, WebSocketUpgrade},
-    extract::{Path, State},
+    extract::{Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
     routing::{get, post},
     Json, Router,
 };
+use serde::Deserialize;
 use serde_json::{json, Value};
 use tokio::sync::broadcast;
 use tracing::{debug, info, warn};
 
 use crate::rpc::{Rpc, RpcError};
 use crate::shard::ShardHandle;
+use crate::store::Store;
 
 /// Shared state handed to each request handler.
 #[derive(Clone)]
@@ -27,6 +29,7 @@ pub struct AppState {
     pub events: broadcast::Sender<String>,
     pub shard: ShardHandle,
     pub rpc: Rpc,
+    pub store: Store,
 }
 
 pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
@@ -40,8 +43,12 @@ pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
         .route("/vendors/:account", get(vendors))
         // Inbound commands (correlated by code / id).
         .route("/link/confirm", post(link_confirm))
+        .route("/link/:account", get(link_lookup))
         .route("/towncrier", post(towncrier_add))
         .route("/towncrier/:id", axum::routing::delete(towncrier_remove))
+        // History, read from SQLite rather than the shard.
+        .route("/history", get(history))
+        .route("/economy", get(economy))
         .with_state(state);
 
     let listener = tokio::net::TcpListener::bind(addr).await?;
@@ -97,7 +104,7 @@ async fn char_by_slot(
 ) -> impl IntoResponse {
     let req_id = st.rpc.next_req_id();
     let cmd = json!({"kind":"char.request","reqId":req_id,"account":account,"slot":slot});
-    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
+    char_reply(&st, st.rpc.call(&st.shard, cmd, &req_id).await).await
 }
 
 async fn char_by_serial(
@@ -105,8 +112,38 @@ async fn char_by_serial(
     Path(serial): Path<String>,
 ) -> impl IntoResponse {
     let req_id = st.rpc.next_req_id();
-    let cmd = json!({"kind":"char.request","reqId":req_id,"serial":serial});
+    let cmd = json!({"kind":"char.request","reqId":req_id,"serial":serial.clone()});
-    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
+    let result = st.rpc.call(&st.shard, cmd, &req_id).await;
+
+    // Resilience: if the shard is unreachable, serve the last cached profile if we have one.
+    if matches!(result, Err(RpcError::NoShard) | Err(RpcError::Timeout)) {
+        if let Ok(Some(cached)) = st.store.get_cached_profile(&serial).await {
+            return (StatusCode::OK, Json(cached));
+        }
+    }
+    char_reply(&st, result).await
+}
+
+/// Caches a successful profile before responding, so a later shard outage can still serve it.
+async fn char_reply(st: &AppState, result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
+    if let Ok(value) = &result {
+        if value.get("kind").and_then(|k| k.as_str()) == Some("char.profile") {
+            if let Some(serial) = value.get("serial").and_then(|s| s.as_str()) {
+                let t = value.get("t").and_then(|v| v.as_i64()).unwrap_or(0);
+                let _ = st
+                    .store
+                    .cache_profile(
+                        serial,
+                        value.get("acct").and_then(|a| a.as_str()),
+                        value.get("name").and_then(|n| n.as_str()),
+                        &value.to_string(),
+                        t,
+                    )
+                    .await;
+            }
+        }
+    }
+    respond(result)
 }
 
 async fn roster(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
@@ -140,7 +177,35 @@ async fn link_confirm(State(st): State<AppState>, Json(body): Json<Value>) -> im
 
     let cmd = json!({"kind":"link.confirm","code":code,"websiteUserId":web_id});
     let code = code.to_string();
-    respond(st.rpc.call(&st.shard, cmd, &code).await)
+    let result = st.rpc.call(&st.shard, cmd, &code).await;
+
+    // Mirror a successful link into the store so events can be attributed without the shard.
+    if let Ok(value) = &result {
+        if value.get("kind").and_then(|k| k.as_str()) == Some("link.ok") {
+            if let Some(account) = value.get("account").and_then(|a| a.as_str()) {
+                let t = value.get("t").and_then(|v| v.as_i64()).unwrap_or(0);
+                let _ = st.store.record_link(account, web_id, t).await;
+            }
+        }
+    }
+    respond(result)
+}
+
+async fn link_lookup(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
+    match st.store.get_link(&account).await {
+        Ok(Some(web_id)) => (
+            StatusCode::OK,
+            Json(json!({"account": account, "websiteUserId": web_id})),
+        ),
+        Ok(None) => (
+            StatusCode::NOT_FOUND,
+            Json(json!({"account": account, "linked": false})),
+        ),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": e.to_string()})),
+        ),
+    }
 }
 
 /// Body: {"id":"n123","lines":["..."],"durationSec":3600}. Correlated on `id`.
@@ -167,6 +232,38 @@ async fn towncrier_remove(
     respond(st.rpc.call(&st.shard, cmd, &id).await)
 }
 
+// ---- history (from SQLite) ----
+
+#[derive(Deserialize)]
+struct HistoryQuery {
+    kind: Option<String>,
+    limit: Option<i64>,
+}
+
+/// Recent events from the store, newest first. `?kind=vendor.sale&limit=50` to filter.
+async fn history(State(st): State<AppState>, Query(q): Query<HistoryQuery>) -> impl IntoResponse {
+    let limit = q.limit.unwrap_or(100);
+    match st.store.recent(q.kind.as_deref(), limit).await {
+        Ok(events) => (StatusCode::OK, Json(json!({"events": events}))),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": e.to_string()})),
+        ),
+    }
+}
+
+/// The money-supply series (economy.supply snapshots), newest first.
+async fn economy(State(st): State<AppState>, Query(q): Query<HistoryQuery>) -> impl IntoResponse {
+    let limit = q.limit.unwrap_or(100);
+    match st.store.economy(limit).await {
+        Ok(series) => (StatusCode::OK, Json(json!({"series": series}))),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": e.to_string()})),
+        ),
+    }
+}
+
 // ---- websocket ----
 
 async fn ws_upgrade(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {