Merge pull request 'feat(sidecar): protocol 10 — five reward forwards (phase 13b)' (#10) from feat/phase-13b-rewards into edge

Reviewed-on: #10

.gitea/workflows/pr-checks.yml

@@ -0,0 +1,99 @@
+# Gate every pull request into `main` on the checks this repository already had
+# and nobody ran automatically.
+#
+# Phases 1 and 3 both wrote `cargo fmt`, `cargo clippy -D warnings` and a test
+# suite, and both ran them BY HAND from a workstation. That is the whole gap
+# this file closes: a guard nothing invokes is a guard whose state nobody knows,
+# and the repository that gets released had nothing gating it at all.
+#
+# Adapted from RunicGateway/installer's pr-checks.yml, which is the other Rust
+# crate in this project and already solved the toolchain-on-a-shared-runner
+# problem. Two differences, both because of where the crate sits:
+#
+#   • The crate is in `sidecar/`, not at the repo root, so every cargo step runs
+#     with that working directory and the cache key reads that lockfile.
+#   • There is no "does a crate exist yet" detection. The installer needed it
+#     because its CI landed before its code; here the code came first.
+#
+# Enforcement (one-time, in the Gitea UI):
+#   Repository Settings → Branches → Branch Protection (rule for `main`)
+#     • Enable Status Check
+#     • Status check patterns:  PR Checks / *
+#   Gitea only lists a context in its dropdown after it has reported once, so let
+#   this run on one PR first; the glob matches without the dropdown and keeps
+#   matching as jobs are added.
+#
+# Scope note: `edge` is gated as well as `main` though this repo has no `edge`
+# branch. Multi-phase work lands there first everywhere else in this project, and
+# gating only the `main` hop would run these checks for the first time at the
+# cutover — the one moment a red build is most expensive to find.
+
+name: PR Checks
+
+on:
+  pull_request:
+    branches: [main, edge]
+
+concurrency:
+  group: pr-checks-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  rust-gates:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    defaults:
+      run:
+        working-directory: sidecar
+    steps:
+      - uses: actions/checkout@v4
+
+      # One job runs all three gates on purpose: installing the toolchain costs
+      # far more than the checks do, so splitting fmt/clippy/test into parallel
+      # jobs would pay that cost three times for no wall-clock win.
+      - name: Install Rust toolchain (rustfmt + clippy)
+        run: |
+          set -euo pipefail
+          SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo"
+          $SUDO apt-get update
+          $SUDO apt-get install -y --no-install-recommends \
+            build-essential curl ca-certificates git
+
+          if ! command -v cargo >/dev/null 2>&1; then
+            curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs \
+              | sh -s -- -y --profile minimal --default-toolchain stable
+          fi
+          echo "${HOME}/.cargo/bin" >> "$GITHUB_PATH"
+          export PATH="${HOME}/.cargo/bin:${PATH}"
+          rustup component add rustfmt clippy
+          cargo --version && cargo fmt --version && cargo clippy --version
+
+      # Keyed on Cargo.lock: dependency builds are reused until a dep actually
+      # changes. A cache miss only makes the run slower, never wrong.
+      - name: Cache cargo registry and build dir
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            sidecar/target
+          key: ${{ runner.os }}-cargo-${{ hashFiles('sidecar/Cargo.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-cargo-
+
+      # Cheapest gate first — parses only, no compile, so a formatting slip fails
+      # in seconds instead of after a full build.
+      - name: cargo fmt --check
+        run: cargo fmt --check
+
+      # --all-targets covers the tests too, which is where most of this crate's
+      # interesting code is. -D warnings makes a lint a failure, so the crate
+      # starts clean at this bar and anything new is a regression from the PR.
+      - name: cargo clippy
+        run: cargo clippy --locked --all-targets -- -D warnings
+
+      # --locked also proves Cargo.lock is in sync with Cargo.toml rather than
+      # letting the build silently update it.
+      - name: cargo test
+        run: cargo test --locked
+

README.md

@@ -0,0 +1,71 @@
+# rust-link
+
+The **sidecar** half of the Runic Gateway bridge for [Rust](https://rust.facepunch.com/). It
+terminates the loopback link from a Rust server's Oxide bridge plugin and exposes the WebSocket +
+REST surface the website consumes.
+
+It is the mirror of [`RunicGateway/link`](https://gitea.whitlocktech.com/RunicGateway/link), which
+does the same job for Ultima Online, and it keeps that bridge's central invariant unchanged:
+
+> **The game server is never reachable from the website.** The plugin dials *out* to this process;
+> this process owns the listener. Only the sidecar is exposed, and only the website's backend talks
+> to it.
+
+```
+Rust server + Oxide  (RunicGateway/Rust-Plugins, C#)
+   │  loopback TCP 127.0.0.1:7799, newline-delimited JSON, bidirectional
+   │  the PLUGIN dials out (the game opens no listening port for us)
+   ▼
+rust-link sidecar  (this repo, Rust)   ← the only network-facing bridge component
+   │  WebSocket (live feed) + REST (point-in-time reads), bearer-token auth
+   ▼
+website backend + module-rust  (RunicGateway/Module-Rust, Node)
+```
+
+## One server, one sidecar
+
+This binary serves **exactly one** Rust game server. A community running six servers runs six
+pairs, each with its own port, database and token; `module-rust` holds six clients and the website
+core never learns there is more than one. Nothing here is multiplexed, and nothing here should
+become multiplexed.
+
+## Build and run
+
+```bash
+cd sidecar
+cargo build --release          # → target/release/rust-link-sidecar
+cargo run                      # info logging; writes sidecar.toml (with a generated token) on first run
+RUST_LOG=debug cargo run       # verbose, including heartbeats
+```
+
+Everything is configured from `sidecar.toml` — nothing is compiled into the binary. **Auth is
+always on**: a blank token is generated and written back on first start, so there is no state in
+which this process listens without one. `--print-config` resolves the configuration and prints it
+as JSON, which is how an installer reads the token back without scraping a log.
+
+See [`sidecar/README.md`](sidecar/README.md) for the configuration reference and the endpoint list.
+
+## The protocol is a contract
+
+The loopback JSON protocol (plugin ↔ sidecar) and this sidecar's HTTP/WS API (sidecar ↔ website)
+are **versioned compatibility contracts**, not build dependencies. `PROTOCOL_VERSION` lives in
+[`sidecar/src/main.rs`](sidecar/src/main.rs); every response carries `X-RustLink-Version`, and a
+client that declares a different one is refused `409` rather than served something it will
+mis-parse.
+
+A version is declared in **three** places and they must agree:
+
+| Where | Repo |
+|---|---|
+| `PROTOCOL_VERSION` | this repo |
+| `overlay.toml` | [`RunicGateway/Rust-Plugins`](https://gitea.whitlocktech.com/RunicGateway/Rust-Plugins) |
+| `module.json` | [`RunicGateway/Module-Rust`](https://gitea.whitlocktech.com/RunicGateway/Module-Rust) |
+
+The canonical spec is
+[`docs/rust-link/`](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/rust-link). If
+you add or change an event or a command, update all three repos **and** the spec in the same
+change.
+
+## Licence
+
+GPL-3.0-or-later. See [LICENSE.md](LICENSE.md).

sidecar/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "rust-link-sidecar"
+version = "0.1.0"
+edition = "2021"
+license = "GPL-3.0-or-later"
+description = "Rust sidecar for the rust-link bridge: terminates the loopback link to a Rust/Oxide game server and exposes WebSocket + REST to the website."
+
+[dependencies]
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "sync", "time", "signal"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+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"] }
+toml = "0.8"
+getrandom = "0.2"
+chrono = { version = "0.4", default-features = false, features = ["std", "clock"] }
+
+[profile.release]
+opt-level = 2

sidecar/README.md

@@ -0,0 +1,116 @@
+# rust-link-sidecar
+
+Configuration reference and endpoint list. For what this component *is*, see the
+[repo README](../README.md).
+
+## Configuration
+
+`sidecar.toml`, resolved in this order: `--config <PATH>`, else `$RUSTLINK_CONFIG`, else
+`./sidecar.toml`. Environment variables override the file; the file overrides the defaults.
+
+| Key | Env | Default | What it is |
+|---|---|---|---|
+| `[game].bind` | `RUSTLINK_GAME_BIND` | `127.0.0.1:7799` | Where the Oxide plugin dials in |
+| `[game].server_id` | `RUSTLINK_SERVER_ID` | *(empty)* | Optional cross-check against the plugin's own `serverId` |
+| `[web].bind` | `RUSTLINK_WEB_BIND` | `127.0.0.1:8090` | Where the website reaches this sidecar |
+| `[web].auth_token` | `RUSTLINK_WEB_TOKEN` | *(generated)* | The shared secret the website presents |
+| `[store].path` | `RUSTLINK_DB_PATH` | `rust-link.db` | SQLite file |
+| `[store].retain_days` | `RUSTLINK_RETAIN_DAYS` | `14` | Days of event history to keep. `0` keeps everything |
+
+Two things about those defaults are load-bearing:
+
+- **`[game].bind` is loopback, and there is no token on that link.** The plugin and the sidecar
+  share a host; `127.0.0.1` *is* the authentication. Binding it to a routable address puts an
+  unauthenticated command channel on the network.
+- **A relative `[store].path` resolves against the directory holding `sidecar.toml`**, not the
+  working directory. A service manager's working directory must not decide where the database
+  lands — on Windows that can be `%SystemRoot%\System32`, or a silently redirected VirtualStore
+  copy.
+
+`[game].server_id` is a **cross-check, not a second source of truth**. The plugin announces its own
+`serverId` and that is the authority; when both are set and they disagree, the sidecar logs the
+disagreement loudly and keeps the plugin's. Two game servers pointed at one sidecar by a copied
+config is the mistake this catches, and it is silent in every other design.
+
+### Reading the token back
+
+```bash
+rust-link-sidecar --print-config
+```
+
+Resolves the configuration exactly as a normal start would — writing the file and generating the
+token if they are missing — and prints it as JSON on stdout, **including the token in clear text**.
+That is the supported way for an installer to obtain it; the alternative is scraping a log.
+
+## Endpoints
+
+Everything except `/health` requires the token, as `Authorization: Bearer <t>`, `X-Api-Key: <t>`,
+or `?token=<t>` (the last so browser WebSocket clients, which cannot set handshake headers, can
+still authenticate). Every response carries `X-RustLink-Version`.
+
+| Route | Backed by | Notes |
+|---|---|---|
+| `GET /health` | — | Unauthenticated, so monitoring can reach it |
+| `GET /server` | store | The last `server.hello`. **`204` when the game has never connected** |
+| `GET /boards` | store | Every board, keyed by kind. `200` with an empty object when there are none |
+| `GET /events?kind=&wipe=&limit=` | store | Newest first; `limit` clamped to 1–1000. For a human |
+| `GET /feed?since=&limit=` | store | **Oldest first, from a cursor.** For a consumer that must not miss a row. Omitting `since` asks where the end is |
+| `GET /status` | plugin (RPC) | A live round trip. `503` with no plugin, `504` on no reply |
+| `GET /ws` | broadcast | The live feed. Sends `ws.hello` on connect |
+
+The split is the point: the store-backed reads answer while the game is off, which is what lets the
+website render a server list during a wipe or a restart. `/status` is the one route that fails when
+the game is down, because "what is it doing right now" has no stale answer worth giving.
+
+`/server` answers `204`, not `200` with a null, when the game has never connected. "We have never
+heard from this server" and "this server reports nothing" are different answers, and a client that
+cannot tell them apart renders a server that does not exist.
+
+## The protocol
+
+Newline-delimited JSON over TCP, both directions. Outbound frames (plugin → sidecar) carry `kind`;
+inbound frames (sidecar → plugin) carry `cmd`. Lines are capped at 1 MiB; an over-long line is
+discarded and the connection stays up.
+
+**This process files a frame by its `type`, and never by its `kind`** — which is what keeps it a
+dumb forwarder while the catalogue grows. Ten new event kinds are no change here at all.
+
+| `type` | Kept | Broadcast | Example |
+|---|---|---|---|
+| `event` | appended to the history | yes | `player.death` |
+| `snapshot` | **replaces** the board of that kind | yes | `players.online` |
+| `reply` | no | no | `server.status`, routed by `reqId` |
+| `control` | no | yes | `pong`, `link.down` |
+
+A frame with no `type` this build knows is **dropped and counted**, never guessed at, and the count
+is on `/health` as `untyped_frames`. The game link has no version handshake, so a plugin and a
+sidecar on different protocol versions show up there and nowhere else.
+
+| Frame | Direction | Purpose |
+|---|---|---|
+| `server.hello` | plugin → sidecar | A board. Sent on **every connect**, not once at server start — this process restarts independently of the game. Carries `serverId`, `bootId` and `wipeId` |
+| `players.online` | plugin → sidecar | The other board: who is connected, re-sent on connect and every 60s |
+| `ping` / `pong` | sidecar → plugin → sidecar | The heartbeat, every 30s. A `pong` is never persisted; it only moves `last_event` |
+| `server.status` | sidecar → plugin → sidecar | The one request/reply verb, correlated by `reqId` |
+| the read path | plugin → sidecar | Presence, deaths, chat, tallies, moderation, the wipe — the catalogue is `PROTOCOL.md` §8.4 |
+
+`bootId` is how a game restart is told apart from a sidecar reconnect — the distinction the event
+system's `reconcile` hangs off later. `wipeId` is how a wipe splits the history instead of ending
+it; the plugin derives it, and every frame carries it.
+
+**History is bounded, boards are not.** `[store].retain_days` (default 14) prunes events hourly; a
+board is one row per kind holding what is true now, and pruning it would make a server the site has
+rendered for weeks look like one that has never connected. The permanent record is the website's.
+
+**The RPC reply timeout (`rpc::REPLY_TIMEOUT`, 10s) is a ceiling every later command budget sits
+under.** Core classifies a budget overrun as retryable unconditionally, because it cannot ask the
+game while the action is still awaiting a socket. An action whose `budgetMs` exceeds this can never
+report `retry: false`.
+
+## Checks
+
+```bash
+cargo fmt --all -- --check
+cargo clippy --all-targets -- -D warnings
+cargo test
+```

sidecar/src/app.rs

@@ -0,0 +1,365 @@
+//! The sidecar itself: everything that happens between "we have a config path" and "we were told
+//! to stop".
+//!
+//! [`run`] is parameterised on the two things a supervisor cares about:
+//!
+//! - `ready` is called once the sidecar is actually up (listener bound, store open). A service
+//!   wrapper reports `Running` to its host there, so a config or bind failure surfaces as a *start*
+//!   failure rather than a service that reports Running and then dies.
+//! - `shutdown` is whatever "stop" means on this host.
+//!
+//! Phase 1 runs in the foreground only; the parameters exist now so that adding a host's own
+//! supervision later is a new module rather than a restructuring of this one.
+
+use std::future::Future;
+use std::sync::atomic::{AtomicI64, AtomicU64};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use tokio::sync::{broadcast, mpsc};
+use tracing::{info, warn};
+
+use crate::{config, game, rpc, store, web};
+
+/// How often the sidecar pings the plugin.
+///
+/// This is the only thing that moves `last_event` on a quiet server, and a Rust server with nobody
+/// on it is very quiet. Without it, "the game has said nothing for six hours" would be
+/// indistinguishable from "the link died six hours ago".
+const HEARTBEAT: Duration = Duration::from_secs(30);
+
+/// What this process does with one frame, decided by the frame's `type` and by nothing else.
+///
+/// Lifting it out of the event loop is not tidiness. This is the whole of protocol 2's filing
+/// rule (PROTOCOL.md §8.1) and the reason a new event kind costs this repository nothing — so it
+/// is worth being a thing that can be asserted about, rather than five arms of a `match` inside a
+/// spawned task that no test can reach.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Filing {
+    /// History. Stored, then broadcast.
+    Persist,
+    /// Current state. Replaces the board of its kind, then broadcast.
+    Board,
+    /// Neither history nor state: broadcast only. The heartbeat, and our own `link.down`.
+    Announce,
+    /// A reply whose caller has already timed out. Storing it would put a point-in-time answer
+    /// into the history as though the game had volunteered it; broadcasting it would show it to a
+    /// website as a live event. Neither is true, so it goes nowhere.
+    Ignore,
+    /// Not something this build knows how to file. Almost always a plugin on a different protocol
+    /// version — the game link has no handshake to catch that earlier.
+    Unfilable,
+}
+
+impl Filing {
+    fn of(frame_type: &str) -> Self {
+        match frame_type {
+            "event" => Filing::Persist,
+            "snapshot" => Filing::Board,
+            "control" => Filing::Announce,
+            "reply" => Filing::Ignore,
+            _ => Filing::Unfilable,
+        }
+    }
+}
+
+/// Runs the sidecar until `shutdown` resolves.
+///
+/// `config_path` is the `--config` argument, or `None` to resolve `$RUSTLINK_CONFIG` and the
+/// default as usual.
+pub async fn run<R, S>(config_path: Option<&str>, ready: R, shutdown: S) -> anyhow::Result<()>
+where
+    R: FnOnce(),
+    S: Future<Output = ()>,
+{
+    info!("rust-link sidecar starting");
+
+    let loaded = config::Config::load(config_path)?;
+    let cfg = loaded.cfg;
+    info!(
+        config = %loaded.path.display(),
+        game = %cfg.game.bind,
+        web = %cfg.web.bind,
+        db = %cfg.store.path,
+        auth = cfg.auth_required(),
+        "configuration loaded"
+    );
+
+    // Game link: events in, commands out.
+    let (event_tx, mut event_rx) = mpsc::unbounded_channel::<game::GameEvent>();
+    let (handle, _bound) = game::serve(&cfg.game.bind, event_tx).await?;
+
+    // Live feed: every game event fans out to all connected website WebSocket clients.
+    let (bcast_tx, _) = broadcast::channel::<String>(1024);
+
+    // Request/reply correlation for REST queries.
+    let rpc = rpc::Rpc::new();
+
+    // Durable store: event history and the server board.
+    let store = store::Store::open(&cfg.store.path).await?;
+
+    // Health/observability state.
+    let started = Instant::now();
+    let last_event = Arc::new(AtomicI64::new(0));
+
+    // Frames this build could not file. Surfaced on /health rather than only in the log, because
+    // the shape it diagnoses — a plugin and a sidecar on different protocol versions — presents to
+    // an operator as "the website shows nothing" and nothing else.
+    let untyped = Arc::new(AtomicU64::new(0));
+
+    // Website-facing HTTP server.
+    let web_state = web::AppState {
+        events: bcast_tx.clone(),
+        game: handle.clone(),
+        rpc: rpc.clone(),
+        store: store.clone(),
+        token: Arc::new(cfg.web.auth_token.clone()),
+        started,
+        last_event: last_event.clone(),
+        untyped: untyped.clone(),
+    };
+    let web_bind = cfg.web.bind.clone();
+    tokio::spawn(async move {
+        if let Err(e) = web::serve(&web_bind, web_state).await {
+            tracing::error!(error = %e, "web server exited");
+        }
+    });
+
+    // The heartbeat. Commands to a disconnected plugin are dropped rather than queued, so this is
+    // safe to fire whether or not anything is connected.
+    let ping_handle = handle.clone();
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(HEARTBEAT);
+        // The first tick fires immediately, which would ping before the plugin has had a chance to
+        // dial in and log a dropped command on every start.
+        tick.tick().await;
+        loop {
+            tick.tick().await;
+            if ping_handle.is_connected().await {
+                ping_handle.send(r#"{"cmd":"ping"}"#.to_string()).await;
+            }
+        }
+    });
+
+    // The 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 filed by its `type`, and by its `type` alone:
+    // that is what keeps this process a dumb forwarder while the catalogue grows (PROTOCOL.md
+    // §8.1). Ten new event kinds are no change here.
+    let configured_server_id = cfg.game.server_id.clone();
+    let feed_tx = bcast_tx.clone();
+    let route_rpc = rpc.clone();
+    let event_store = store.clone();
+    let last_event_ts = last_event.clone();
+    let untyped_count = untyped.clone();
+    tokio::spawn(async move {
+        let mut total: u64 = 0;
+        let mut warned_untyped = false;
+
+        while let Some(ev) = event_rx.recv().await {
+            // Any line from the plugin — a pong included — is a sign of life.
+            last_event_ts.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
+
+            if route_rpc.try_route(&ev.value).await {
+                continue; // consumed as a reply
+            }
+
+            let line = ev.value.to_string();
+            let frame_type = ev
+                .value
+                .get("type")
+                .and_then(|v| v.as_str())
+                .unwrap_or("")
+                .to_string();
+
+            match Filing::of(&frame_type) {
+                Filing::Persist => {
+                    total += 1;
+                    tracing::debug!(kind = %ev.kind, n = total, "{}", line);
+
+                    if let Err(e) = event_store
+                        .insert_event(
+                            frame_t(&ev.value),
+                            &ev.kind,
+                            field(&ev.value, "serverId"),
+                            field(&ev.value, "wipeId"),
+                            &line,
+                        )
+                        .await
+                    {
+                        store::warn_write("event persist failed", &e);
+                    }
+                }
+                Filing::Board => {
+                    // A board REPLACES rather than appends. Filing one as history is the mistake
+                    // `type` exists to prevent, and it is invisible until somebody wonders why the
+                    // presence board has four thousand rows.
+                    if ev.kind == store::SERVER_BOARD {
+                        check_server_id(&configured_server_id, &ev.value);
+                        info!(kind = %ev.kind, "{}", line);
+                    } else {
+                        tracing::debug!(kind = %ev.kind, "board {}", line);
+                    }
+
+                    if let Err(e) = event_store
+                        .put_board(&ev.kind, frame_t(&ev.value), &line)
+                        .await
+                    {
+                        store::warn_write("board write failed", &e);
+                    }
+                }
+                Filing::Announce => {
+                    if ev.kind == "link.down" {
+                        info!("game link down");
+                    }
+                }
+                Filing::Ignore => {
+                    tracing::debug!(kind = %ev.kind, "unrouted reply discarded");
+                    continue;
+                }
+                Filing::Unfilable => {
+                    untyped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
+                    if !warned_untyped {
+                        warned_untyped = true;
+                        warn!(
+                            kind = %ev.kind,
+                            frame_type = %frame_type,
+                            "frame with no usable `type`; dropping. The plugin and this sidecar                              are almost certainly on different protocol versions — the game link                              has no version handshake, so this is where that shows up. Counted on                              /health as untyped_frames."
+                        );
+                    }
+
+                    continue; // not stored, and not broadcast: nobody downstream can file it either
+                }
+            }
+
+            // Broadcast after persisting, so a client that reacts by reading the store cannot
+            // beat its own event there. `send` fails only when nobody is subscribed.
+            let _ = feed_tx.send(line);
+        }
+    });
+
+    // Retention. Hourly, and once at startup so a store that grew while an operator had this
+    // turned off does not wait an hour to shrink.
+    let prune_store = store.clone();
+    let retain_days = cfg.store.retain_days;
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(Duration::from_secs(3600));
+        loop {
+            tick.tick().await;
+            if let Err(e) = prune_store.prune(retain_days).await {
+                store::warn_write("prune failed", &e);
+            }
+        }
+    });
+
+    ready();
+    info!("rust-link sidecar ready");
+
+    shutdown.await;
+    info!("shutting down");
+    Ok(())
+}
+
+/// The plugin's `serverId` is the authority; `[game].server_id` is a cross-check. A disagreement is
+/// almost always two game servers pointed at one sidecar by a copied config, which is silent in
+/// every other design and produces one server's history under another's name.
+fn check_server_id(configured: &str, hello: &serde_json::Value) {
+    if configured.is_empty() {
+        return;
+    }
+    let announced = hello.get("serverId").and_then(|v| v.as_str()).unwrap_or("");
+    if !announced.is_empty() && announced != configured {
+        warn!(
+            configured,
+            announced,
+            "the connected plugin announces a different serverId than this sidecar is configured \
+             for; keeping the plugin's. Two servers sharing one sidecar is the usual cause."
+        );
+    }
+}
+
+/// A frame's own timestamp, falling back to ours. The plugin stamps `t` at the moment the world was
+/// read, which is earlier than the moment we saw the line and is the one worth keeping.
+fn frame_t(value: &serde_json::Value) -> i64 {
+    value
+        .get("t")
+        .and_then(|v| v.as_i64())
+        .unwrap_or_else(now_ms)
+}
+
+/// Lifts an optional string field out of a frame, for the columns the store indexes on.
+///
+/// Absent and empty are the same answer here — `None` — because the plugin omits a field it cannot
+/// answer (a server that has never saved has no wipe) and an empty string in an indexed column
+/// would group every such row together as though they shared something.
+fn field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> {
+    value
+        .get(key)
+        .and_then(|v| v.as_str())
+        .filter(|s| !s.is_empty())
+}
+
+fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn a_frames_own_timestamp_wins_over_ours() {
+        assert_eq!(
+            frame_t(&json!({"t": 1_700_000_000_000i64})),
+            1_700_000_000_000
+        );
+    }
+
+    /// The filing rule, which is the whole of protocol 2 in this file.
+    #[test]
+    fn a_frame_is_filed_by_its_type_and_by_nothing_else() {
+        assert_eq!(Filing::of("event"), Filing::Persist);
+        assert_eq!(Filing::of("snapshot"), Filing::Board);
+        assert_eq!(Filing::of("control"), Filing::Announce);
+        assert_eq!(Filing::of("reply"), Filing::Ignore);
+    }
+
+    /// The two ways a frame arrives unfilable, and the reason neither may be guessed at.
+    ///
+    /// Defaulting an absent `type` to `event` would file a BOARD as history — the presence board
+    /// appended a few thousand times, which nothing reports and nobody notices until they wonder
+    /// why the database is large. A frame nobody can file is dropped loudly instead, and counted
+    /// where an operator can see it.
+    #[test]
+    fn an_absent_or_unknown_type_is_never_guessed_at() {
+        assert_eq!(Filing::of(""), Filing::Unfilable);
+        assert_eq!(Filing::of("events"), Filing::Unfilable);
+        assert_eq!(Filing::of("Event"), Filing::Unfilable);
+        assert_eq!(Filing::of("boards"), Filing::Unfilable);
+    }
+
+    /// The columns the store indexes on come out of the frame here, and "absent" and "empty" have
+    /// to be the same answer: a server that has never saved sends no `wipeId`, and an empty string
+    /// in an indexed column would group every such row together as though they shared a wipe.
+    #[test]
+    fn an_empty_string_is_not_a_value() {
+        let frame = json!({"serverId": "main", "wipeId": "", "other": 3});
+        assert_eq!(field(&frame, "serverId"), Some("main"));
+        assert_eq!(field(&frame, "wipeId"), None);
+        assert_eq!(field(&frame, "missing"), None);
+        assert_eq!(field(&frame, "other"), None);
+    }
+
+    /// A frame with no `t` must still land with a usable timestamp rather than at the epoch, or
+    /// every un-stamped event sorts to the beginning of history forever.
+    #[test]
+    fn a_frame_without_a_timestamp_gets_one() {
+        assert!(frame_t(&json!({"kind": "x"})) > 1_600_000_000_000);
+        // A non-numeric `t` is a malformed frame, not a zero.
+        assert!(frame_t(&json!({"t": "nope"})) > 1_600_000_000_000);
+    }
+}

sidecar/src/cli.rs

@@ -0,0 +1,170 @@
+//! Command-line surface.
+//!
+//! The sidecar is configured by file and environment (see [`crate::config`]); this is deliberately
+//! not a second configuration mechanism. It exists so the binary can be *driven by an installer*
+//! rather than only by a human reading its logs:
+//!
+//! - `--print-config` resolves the configuration exactly as a normal start would — including
+//!   generating the auth token on first run — and prints it as JSON on stdout. That is the
+//!   supported way to obtain the token for the website's Admin -> Modules -> Rust form.
+//! - `--config <PATH>` names the config file without having to export `$RUSTLINK_CONFIG`, so a
+//!   diagnostic run can point at an installed config from any working directory.
+//!
+//! Hand-rolled rather than pulled from a crate: four flags, no subcommands, no completions.
+
+/// What this invocation should do. Everything except `Run` prints and exits.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Mode {
+    /// Normal operation: bind the game listener and the web server.
+    Run,
+    /// Resolve config, print it as JSON, exit.
+    PrintConfig,
+    Help,
+    Version,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub struct Cli {
+    pub mode: Mode,
+    /// `--config <PATH>`, which outranks `$RUSTLINK_CONFIG`.
+    pub config: Option<String>,
+}
+
+pub const USAGE: &str = "\
+rust-link sidecar — bridges one Rust game server to the Runic Gateway website.
+
+Usage: rust-link-sidecar [OPTIONS]
+
+Options:
+      --print-config     Resolve the configuration, print it as JSON, and exit.
+                         Runs first-run setup like a normal start does: if the
+                         config file is missing it is written, and a blank auth
+                         token is generated and saved. The JSON CONTAINS THE
+                         AUTH TOKEN in clear text.
+      --config <PATH>    Path to sidecar.toml. Overrides $RUSTLINK_CONFIG;
+                         defaults to ./sidecar.toml.
+  -V, --version          Print the sidecar and protocol versions and exit.
+  -h, --help             Print this help and exit.
+
+Configuration lives in sidecar.toml; environment variables override the file:
+  RUSTLINK_CONFIG, RUSTLINK_GAME_BIND, RUSTLINK_WEB_BIND, RUSTLINK_WEB_TOKEN,
+  RUSTLINK_DB_PATH, RUSTLINK_SERVER_ID
+";
+
+/// Parses arguments **without** the program name.
+///
+/// Returns the message to print on stderr when the arguments are unusable; the caller exits `2`.
+pub fn parse<I: IntoIterator<Item = String>>(args: I) -> Result<Cli, String> {
+    let mut mode = Mode::Run;
+    let mut config = None;
+    let mut it = args.into_iter();
+
+    while let Some(arg) = it.next() {
+        match arg.as_str() {
+            "--print-config" => mode = Mode::PrintConfig,
+            "-h" | "--help" => {
+                return Ok(Cli {
+                    mode: Mode::Help,
+                    config,
+                })
+            }
+            "-V" | "--version" => {
+                return Ok(Cli {
+                    mode: Mode::Version,
+                    config,
+                })
+            }
+            "--config" => {
+                // `--config` with nothing after it would otherwise silently fall through and start
+                // the sidecar against the default config — the opposite of what was asked for.
+                let path = it
+                    .next()
+                    .ok_or_else(|| "--config requires a path".to_string())?;
+                config = Some(path);
+            }
+            _ => match arg.strip_prefix("--config=") {
+                Some("") => return Err("--config requires a path".into()),
+                Some(path) => config = Some(path.to_string()),
+                None => return Err(format!("unrecognized argument: {arg}")),
+            },
+        }
+    }
+
+    Ok(Cli { mode, config })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn parse_str(args: &[&str]) -> Result<Cli, String> {
+        parse(args.iter().map(|s| s.to_string()))
+    }
+
+    #[test]
+    fn no_arguments_runs_the_sidecar() {
+        let cli = parse_str(&[]).unwrap();
+        assert_eq!(cli.mode, Mode::Run);
+        assert_eq!(cli.config, None);
+    }
+
+    #[test]
+    fn print_config_is_recognized() {
+        assert_eq!(
+            parse_str(&["--print-config"]).unwrap().mode,
+            Mode::PrintConfig
+        );
+    }
+
+    #[test]
+    fn config_accepts_both_spellings() {
+        let spaced = parse_str(&["--config", "/etc/runicgateway/sidecar.toml"]).unwrap();
+        let equals = parse_str(&["--config=/etc/runicgateway/sidecar.toml"]).unwrap();
+        assert_eq!(
+            spaced.config.as_deref(),
+            Some("/etc/runicgateway/sidecar.toml")
+        );
+        assert_eq!(spaced, equals);
+    }
+
+    #[test]
+    fn config_combines_with_print_config() {
+        let cli = parse_str(&["--config", "c.toml", "--print-config"]).unwrap();
+        assert_eq!(cli.mode, Mode::PrintConfig);
+        assert_eq!(cli.config.as_deref(), Some("c.toml"));
+    }
+
+    #[test]
+    fn a_path_is_never_swallowed_as_a_flag() {
+        // `--config --print-config` takes the next token as the path, wrong as that path is. The
+        // alternative — treating it as a missing value — guesses at intent.
+        let cli = parse_str(&["--config", "--print-config"]).unwrap();
+        assert_eq!(cli.mode, Mode::Run);
+        assert_eq!(cli.config.as_deref(), Some("--print-config"));
+    }
+
+    #[test]
+    fn config_without_a_value_is_an_error() {
+        assert!(parse_str(&["--config"]).is_err());
+        assert!(parse_str(&["--config="]).is_err());
+    }
+
+    #[test]
+    fn unknown_arguments_are_rejected() {
+        // Silently ignoring a typo'd flag would start a sidecar that is not what was asked for.
+        let err = parse_str(&["--pirnt-config"]).unwrap_err();
+        assert!(err.contains("--pirnt-config"), "{err}");
+        assert!(parse_str(&["/etc/runicgateway/sidecar.toml"]).is_err());
+    }
+
+    #[test]
+    fn help_and_version_win_immediately() {
+        assert_eq!(parse_str(&["--help", "--bogus"]).unwrap().mode, Mode::Help);
+        assert_eq!(parse_str(&["-h"]).unwrap().mode, Mode::Help);
+        assert_eq!(
+            parse_str(&["--version", "--bogus"]).unwrap().mode,
+            Mode::Version
+        );
+        assert_eq!(parse_str(&["-V"]).unwrap().mode, Mode::Version);
+    }
+}

sidecar/src/config.rs

@@ -0,0 +1,452 @@
+//! Runtime configuration, loaded from an external file — nothing here is compiled into the binary.
+//!
+//! Precedence: environment variables override the file, the file overrides built-in defaults. On
+//! first run, if the file is absent, a default one is written with a freshly generated auth token,
+//! so the sidecar is secured out of the box and the operator just copies the token to the website.
+//!
+//! File path: `--config <PATH>`, else `$RUSTLINK_CONFIG`, else `sidecar.toml` in the working
+//! directory.
+//!
+//! **Paths are anchored to the config file, not the working directory.** A relative `[store].path`
+//! resolves against the directory holding `sidecar.toml`, so a service started with
+//! `RUSTLINK_CONFIG=/etc/runicgateway/rust-main.toml` keeps its database beside its config instead
+//! of wherever the service manager happened to set the working directory.
+//!
+//! # `server_id`, and why it is here rather than only in the plugin
+//!
+//! R8 makes the platform multi-server: one sidecar per game server, and every row the module
+//! stores carries the server it came from. The plugin declares its own `serverId` in `server.hello`
+//! and that is the authority. This setting is a **cross-check**, not a second source of truth: when
+//! both are set and they disagree, the sidecar logs the disagreement loudly and keeps the plugin's.
+//! Two servers pointed at one sidecar by a copy-pasted config is the mistake this catches, and it
+//! is silent in every other design.
+
+use std::env;
+use std::fs;
+use std::path::{Component, Path, PathBuf};
+
+use serde::Deserialize;
+use serde_json::json;
+use tracing::info;
+
+use crate::PROTOCOL_VERSION;
+
+#[derive(Debug, Default, Deserialize)]
+pub struct Config {
+    #[serde(default)]
+    pub game: GameCfg,
+    #[serde(default)]
+    pub web: WebCfg,
+    #[serde(default)]
+    pub store: StoreCfg,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct GameCfg {
+    #[serde(default = "default_game_bind")]
+    pub bind: String,
+    /// Optional cross-check against the `serverId` the plugin announces. See the module docs.
+    #[serde(default)]
+    pub server_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct WebCfg {
+    #[serde(default = "default_web_bind")]
+    pub bind: String,
+    /// Shared secret the website must present. Never empty in practice — `Config::load` generates
+    /// and persists one when it finds none, so the web surface is authenticated from first boot.
+    #[serde(default)]
+    pub auth_token: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct StoreCfg {
+    #[serde(default = "default_db_path")]
+    pub path: String,
+    /// How many days of event history to keep. `0` keeps everything.
+    ///
+    /// The store sits on a game host, and protocol 2 gave it a catalogue that produces real volume
+    /// — every death, every chat line, every connect. The *permanent* record is the website's:
+    /// per-wipe rollups in the module's own tables (R12). So this bounds the sidecar's copy, and
+    /// the default is generous enough that nobody needs to think about it and small enough that a
+    /// busy month is not a wipe-day outage.
+    #[serde(default = "default_retain_days")]
+    pub retain_days: i64,
+}
+
+/// A loaded configuration plus what loading it *did* — an installer re-running the binary needs to
+/// distinguish "read an existing install" from "provisioned a new one", and it cannot tell from the
+/// values alone.
+#[derive(Debug)]
+pub struct Loaded {
+    pub cfg: Config,
+    /// Absolute path of the config file that was read or written.
+    pub path: PathBuf,
+    /// The config file did not exist and was created by this run.
+    pub config_created: bool,
+    /// No usable token was configured, so one was generated and saved.
+    pub token_generated: bool,
+}
+
+fn default_game_bind() -> String {
+    "127.0.0.1:7799".into()
+}
+fn default_web_bind() -> String {
+    "127.0.0.1:8090".into()
+}
+fn default_db_path() -> String {
+    "rust-link.db".into()
+}
+fn default_retain_days() -> i64 {
+    14
+}
+
+impl Default for GameCfg {
+    fn default() -> Self {
+        Self {
+            bind: default_game_bind(),
+            server_id: String::new(),
+        }
+    }
+}
+impl Default for WebCfg {
+    fn default() -> Self {
+        Self {
+            bind: default_web_bind(),
+            auth_token: String::new(),
+        }
+    }
+}
+impl Default for StoreCfg {
+    fn default() -> Self {
+        Self {
+            path: default_db_path(),
+            retain_days: default_retain_days(),
+        }
+    }
+}
+
+impl Config {
+    /// Which config file this invocation will use: `--config`, else `$RUSTLINK_CONFIG`, else
+    /// `sidecar.toml` beside the working directory. Always returned absolute, so every later
+    /// message names a path the operator can act on.
+    pub fn resolve_path(cli_override: Option<&str>) -> PathBuf {
+        let raw = cli_override
+            .map(str::to_string)
+            .or_else(|| env::var("RUSTLINK_CONFIG").ok())
+            .unwrap_or_else(|| "sidecar.toml".into());
+        absolutize(PathBuf::from(raw))
+    }
+
+    pub fn load(cli_override: Option<&str>) -> anyhow::Result<Loaded> {
+        let path = Self::resolve_path(cli_override);
+        let existed = path.exists();
+
+        let mut cfg: Config = if existed {
+            let text = fs::read_to_string(&path)?;
+            toml::from_str(&text)?
+        } else {
+            Config::default()
+        };
+
+        cfg.apply_env();
+
+        // Authentication is always on. A blank token is never allowed — if none is set (fresh
+        // install, or someone cleared it), generate one, save it, and continue. This keeps setup
+        // effortless while making it impossible to accidentally run with auth off.
+        let token_generated = cfg.web.auth_token.trim().is_empty();
+        if token_generated {
+            let token = generate_token();
+
+            if existed {
+                persist_token(&path, &token)?;
+            } else {
+                // The parent may not exist yet when an installer points at a fresh
+                // /etc/runicgateway; failing here would mean "run me again after mkdir".
+                create_parent_dir(&path)?;
+                fs::write(&path, default_file(&token))?;
+            }
+
+            cfg.web.auth_token = token.clone();
+
+            info!("No auth token configured.");
+            info!("Generated new token: {}", token);
+            info!("Saved to {}. Authentication is on.", path.display());
+        }
+
+        cfg.anchor_store_path(&path);
+
+        Ok(Loaded {
+            cfg,
+            path,
+            config_created: !existed,
+            token_generated,
+        })
+    }
+
+    /// Environment overrides, so a deployment can set secrets without editing the file.
+    fn apply_env(&mut self) {
+        if let Ok(v) = env::var("RUSTLINK_GAME_BIND") {
+            self.game.bind = v;
+        }
+        if let Ok(v) = env::var("RUSTLINK_SERVER_ID") {
+            self.game.server_id = v;
+        }
+        if let Ok(v) = env::var("RUSTLINK_WEB_BIND") {
+            self.web.bind = v;
+        }
+        if let Ok(v) = env::var("RUSTLINK_WEB_TOKEN") {
+            self.web.auth_token = v;
+        }
+        if let Ok(v) = env::var("RUSTLINK_DB_PATH") {
+            self.store.path = v;
+        }
+        if let Ok(v) = env::var("RUSTLINK_RETAIN_DAYS") {
+            // A malformed value is ignored rather than fatal: this reaches the process as a panel
+            // variable somebody typed (R22), and refusing to start over a stray character would
+            // take the bridge down for a setting that has a perfectly good default.
+            match v.trim().parse::<i64>() {
+                Ok(days) if days >= 0 => self.store.retain_days = days,
+                _ => tracing::warn!(value = %v, "ignoring an unreadable RUSTLINK_RETAIN_DAYS"),
+            }
+        }
+    }
+
+    /// Resolves `[store].path` against the config file's directory (see the module docs). Absolute
+    /// paths and SQLite's non-filesystem spellings are left exactly as written.
+    fn anchor_store_path(&mut self, config_path: &Path) {
+        if is_sqlite_special(&self.store.path) {
+            return;
+        }
+        let raw = PathBuf::from(&self.store.path);
+        let anchored = if raw.is_absolute() {
+            raw
+        } else {
+            config_dir(config_path).join(raw)
+        };
+        self.store.path = absolutize(anchored).to_string_lossy().into_owned();
+    }
+
+    pub fn auth_required(&self) -> bool {
+        // Always true now — load() guarantees a non-empty token.
+        !self.web.auth_token.is_empty()
+    }
+}
+
+/// The `--print-config` document: everything an installer needs to register this sidecar with a
+/// website, in one non-interactive read.
+///
+/// **This includes the auth token in clear text**, which is the point: the manual token hunt is the
+/// largest "I installed it and nothing happened" failure mode. The caller prints it to stdout and
+/// starts no log subscriber, so the document is the whole output.
+pub fn describe(loaded: &Loaded) -> serde_json::Value {
+    json!({
+        "component": "rust-link-sidecar",
+        "version": env!("CARGO_PKG_VERSION"),
+        "protocol": PROTOCOL_VERSION,
+        "config_path": loaded.path.to_string_lossy(),
+        "config_created": loaded.config_created,
+        "token_generated": loaded.token_generated,
+        "game": {
+            "bind": loaded.cfg.game.bind,
+            "server_id": loaded.cfg.game.server_id,
+        },
+        "web": {
+            "bind": loaded.cfg.web.bind,
+            "ws_path": crate::web::WS_PATH,
+            "auth_required": loaded.cfg.auth_required(),
+            "auth_token": loaded.cfg.web.auth_token,
+        },
+        "store": { "path": loaded.cfg.store.path },
+    })
+}
+
+/// Directory holding the config file. A bare `sidecar.toml` has no parent component, which would
+/// join into an empty base — treat it as the current directory.
+fn config_dir(config_path: &Path) -> PathBuf {
+    match config_path.parent() {
+        Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
+        _ => PathBuf::from("."),
+    }
+}
+
+/// Prefixes the working directory onto a relative path, then drops the `.` components that
+/// joining leaves behind — cosmetic, but these paths are printed and pasted into service units.
+fn absolutize(p: PathBuf) -> PathBuf {
+    let joined = if p.is_absolute() {
+        p
+    } else {
+        match env::current_dir() {
+            Ok(cwd) => cwd.join(p),
+            Err(_) => p,
+        }
+    };
+    let cleaned: PathBuf = joined
+        .components()
+        .filter(|c| !matches!(c, Component::CurDir))
+        .collect();
+    if cleaned.as_os_str().is_empty() {
+        joined
+    } else {
+        cleaned
+    }
+}
+
+/// `:memory:` and `file:` URIs are instructions to SQLite, not paths on disk. Anchoring them to a
+/// directory would turn a working in-memory store into an attempt to create a file called
+/// `:memory:` — which Windows cannot even name.
+fn is_sqlite_special(path: &str) -> bool {
+    path == ":memory:" || path.starts_with("file:")
+}
+
+fn create_parent_dir(path: &Path) -> anyhow::Result<()> {
+    if let Some(dir) = path.parent() {
+        if !dir.as_os_str().is_empty() && !dir.exists() {
+            fs::create_dir_all(dir)?;
+        }
+    }
+    Ok(())
+}
+
+/// Rewrites the `auth_token` line in an existing config file, preserving everything else. Falls
+/// back to inserting it under `[web]`, or appending a `[web]` section, if the key is absent.
+fn persist_token(path: &Path, token: &str) -> anyhow::Result<()> {
+    let text = fs::read_to_string(path)?;
+    let line = format!("auth_token = \"{token}\"");
+
+    if text
+        .lines()
+        .any(|l| l.trim_start().starts_with("auth_token"))
+    {
+        let out: String = text
+            .lines()
+            .map(|l| {
+                if l.trim_start().starts_with("auth_token") {
+                    line.clone()
+                } else {
+                    l.to_string()
+                }
+            })
+            .collect::<Vec<_>>()
+            .join("\n");
+        fs::write(path, out + "\n")?;
+    } else if text.lines().any(|l| l.trim() == "[web]") {
+        let out: String = text
+            .lines()
+            .flat_map(|l| {
+                if l.trim() == "[web]" {
+                    vec![l.to_string(), line.clone()]
+                } else {
+                    vec![l.to_string()]
+                }
+            })
+            .collect::<Vec<_>>()
+            .join("\n");
+        fs::write(path, out + "\n")?;
+    } else {
+        fs::write(path, format!("{text}\n[web]\n{line}\n"))?;
+    }
+
+    Ok(())
+}
+
+fn generate_token() -> String {
+    let mut buf = [0u8; 24];
+    // OS randomness; falls back to a time-seeded token only if the OS RNG is unavailable.
+    if getrandom::getrandom(&mut buf).is_err() {
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0);
+        return format!("insecure-fallback-{nanos:x}");
+    }
+    buf.iter().map(|b| format!("{b:02x}")).collect()
+}
+
+fn default_file(token: &str) -> String {
+    format!(
+        r#"# rust-link sidecar configuration.
+#
+# One sidecar serves one Rust game server. A community running six servers runs
+# six of these, each with its own port, its own database and its own token.
+#
+# Environment variables override every value here.
+
+[game]
+# Where the Oxide bridge plugin dials in. The plugin is the client; this is the
+# listener, which is why the game server itself opens no extra port.
+bind = "127.0.0.1:7799"
+
+# Optional. If set, it is cross-checked against the serverId the plugin
+# announces in server.hello; a disagreement is logged and the plugin wins.
+server_id = ""
+
+[web]
+# Where the website reaches this sidecar. Bind to a LAN or public address only
+# behind TLS and a firewall — the token below is the only thing guarding it.
+bind = "127.0.0.1:8090"
+
+# Generated on first run. Paste it into the website's Rust server form. It is
+# write-only there: the site never shows it back.
+auth_token = "{token}"
+
+[store]
+# Relative paths resolve against the directory holding THIS FILE, not the
+# working directory of whatever started the process.
+path = "rust-link.db"
+
+# How many days of event history to keep. 0 keeps everything.
+#
+# The permanent record is the website's — it holds per-wipe rollups that survive
+# a wipe. This database is the recent copy the site reads to catch up, and it
+# lives on the game host, so it is bounded.
+retain_days = 14
+"#
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_generated_token_is_48_hex_characters() {
+        let t = generate_token();
+        assert!(!t.starts_with("insecure-fallback-"), "OS RNG unavailable");
+        assert_eq!(t.len(), 48);
+        assert!(t.chars().all(|c| c.is_ascii_hexdigit()));
+    }
+
+    #[test]
+    fn sqlite_special_paths_are_not_anchored() {
+        let mut cfg = Config::default();
+        cfg.store.path = ":memory:".into();
+        cfg.anchor_store_path(Path::new("/etc/runicgateway/sidecar.toml"));
+        assert_eq!(cfg.store.path, ":memory:");
+    }
+
+    /// The whole point of anchoring: a service manager's working directory must not decide where
+    /// the database lands.
+    #[test]
+    fn a_relative_store_path_anchors_to_the_config_directory() {
+        let mut cfg = Config::default();
+        cfg.store.path = "rust-link.db".into();
+        let config_path = absolutize(PathBuf::from("cfgdir/sidecar.toml"));
+        cfg.anchor_store_path(&config_path);
+
+        let expected = config_path.parent().unwrap().join("rust-link.db");
+        assert_eq!(PathBuf::from(&cfg.store.path), expected);
+    }
+
+    /// The written default must be loadable by the loader that wrote it — a template with a typo
+    /// in it fails on the second start, not the first.
+    #[test]
+    fn the_default_file_round_trips() {
+        let cfg: Config = toml::from_str(&default_file("deadbeef")).unwrap();
+        assert_eq!(cfg.web.auth_token, "deadbeef");
+        assert_eq!(cfg.game.bind, default_game_bind());
+        assert_eq!(cfg.store.path, default_db_path());
+        assert_eq!(cfg.game.server_id, "");
+    }
+}

sidecar/src/game.rs

@@ -0,0 +1,418 @@
+//! The loopback link to the Rust server's Oxide bridge plugin.
+//!
+//! The plugin is the TCP *client*: it dials out to us. So the sidecar owns the listener, and the
+//! plugin's outbound socket is the only thing that ever connects. This is the whole reason the game
+//! is never directly reachable from the website — it exposes no port for us.
+//!
+//! Framing is newline-delimited JSON, bidirectional: the plugin sends events (`kind`), we send
+//! commands (`cmd`). We accept one plugin connection at a time and re-accept when it drops (the
+//! plugin reconnects on its own, with a bounded backoff).
+//!
+//! **Loopback is the trust boundary.** There is no token on this link, exactly as on the ServUO
+//! bridge: the plugin and the sidecar share a host, and the address to bind is `127.0.0.1`. Binding
+//! `[game].bind` to anything routable puts an unauthenticated command channel on the network, and
+//! the operator documentation says so in as many words.
+//!
+//! Inbound lines are **capped** (see [`MAX_INBOUND_LINE_BYTES`]) from the start rather than after
+//! the first large frame arrives: an unbounded `read_line` facing a peer that will one day send a
+//! map image is a memory-exhaustion shape we would be inventing ourselves.
+
+use std::sync::Arc;
+
+use serde_json::Value;
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::TcpListener;
+use tokio::sync::{mpsc, Mutex};
+use tracing::{info, warn};
+
+/// The longest line the sidecar will accept from the plugin, in bytes.
+///
+/// Over-long lines are **discarded, not buffered**, and the connection stays up: a single malformed
+/// frame is not a reason to tear down a link that live events are flowing over. A dropped reply
+/// simply times out on the caller's side and is re-requested.
+pub const MAX_INBOUND_LINE_BYTES: usize = 1024 * 1024;
+
+/// What one read off the plugin socket produced.
+#[derive(Debug)]
+enum Line {
+    /// A complete line, within the cap.
+    Complete(String),
+    /// A line that ran past the cap. Carries how many bytes were thrown away, for the log.
+    TooLong(usize),
+    /// The plugin closed the connection.
+    Eof,
+}
+
+/// A cancel-safe, capped, newline-delimited reader.
+///
+/// Every piece of state that must survive a partial read lives here rather than in a local, because
+/// this is polled inside a `tokio::select!`: the loop below drops the future whenever a command
+/// wins the race, and a `discarding` flag or a half-filled buffer held in a local would be lost
+/// with it. Losing the buffer corrupts the *next* line; losing `discarding` turns the tail of an
+/// over-long line into a line of its own. Both are silent.
+///
+/// The only await point is `fill_buf`, and nothing is consumed until after it returns, so a
+/// cancellation between the two can lose at most the wakeup.
+#[derive(Default)]
+struct LineReader {
+    buf: Vec<u8>,
+    discarding: bool,
+    discarded: usize,
+}
+
+impl LineReader {
+    async fn next<R: AsyncBufRead + Unpin>(&mut self, reader: &mut R) -> std::io::Result<Line> {
+        loop {
+            let consumed;
+            let outcome;
+
+            {
+                let available = reader.fill_buf().await?;
+
+                if available.is_empty() {
+                    return Ok(Line::Eof);
+                }
+
+                match available.iter().position(|&b| b == b'\n') {
+                    Some(at) => {
+                        consumed = at + 1;
+
+                        if self.discarding {
+                            // The tail of a line we already gave up on. Swallow it, terminator
+                            // included, and report the size once.
+                            self.discarded += at;
+                            let total = self.discarded;
+                            self.discarding = false;
+                            self.discarded = 0;
+                            outcome = Some(Line::TooLong(total));
+                        } else if self.buf.len() + at > MAX_INBOUND_LINE_BYTES {
+                            // The cap is reached only now, on the chunk that also holds the
+                            // terminator — so there is nothing left to discard.
+                            let total = self.buf.len() + at;
+                            self.buf.clear();
+                            outcome = Some(Line::TooLong(total));
+                        } else {
+                            self.buf.extend_from_slice(&available[..at]);
+                            let line = String::from_utf8_lossy(&self.buf).into_owned();
+                            self.buf.clear();
+                            outcome = Some(Line::Complete(line));
+                        }
+                    }
+                    None => {
+                        consumed = available.len();
+
+                        if self.discarding {
+                            self.discarded += consumed;
+                        } else if self.buf.len() + consumed > MAX_INBOUND_LINE_BYTES {
+                            // Refuse rather than buffer: this is the whole point of the cap.
+                            // Everything up to the next newline is now dropped on the floor.
+                            self.discarded = self.buf.len() + consumed;
+                            self.buf.clear();
+                            self.discarding = true;
+                        } else {
+                            self.buf.extend_from_slice(available);
+                        }
+
+                        outcome = None;
+                    }
+                }
+            }
+
+            reader.consume(consumed);
+
+            if let Some(line) = outcome {
+                return Ok(line);
+            }
+        }
+    }
+}
+
+/// An event line received from the plugin, parsed. `kind` is lifted out for routing.
+#[derive(Debug, Clone)]
+pub struct GameEvent {
+    pub kind: String,
+    pub value: Value,
+}
+
+/// A handle for sending command lines to the plugin. Cloneable and cheap.
+///
+/// Commands are dropped (with a warning) when no plugin is connected, rather than buffered: a
+/// website query that arrives during an outage should fail fast and be retried, not silently queue
+/// behind a reconnect. Live *events* are what must survive an outage, and those the plugin buffers
+/// on its side.
+#[derive(Clone)]
+pub struct GameHandle {
+    tx: Arc<Mutex<Option<mpsc::UnboundedSender<String>>>>,
+}
+
+impl GameHandle {
+    fn new() -> Self {
+        Self {
+            tx: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    async fn set(&self, sender: Option<mpsc::UnboundedSender<String>>) {
+        *self.tx.lock().await = sender;
+    }
+
+    /// Send one command line (a complete JSON object, no newline — we add the frame delimiter).
+    /// Returns false if no plugin is currently connected.
+    pub async fn send(&self, line: String) -> bool {
+        let guard = self.tx.lock().await;
+        match guard.as_ref() {
+            Some(sender) => sender.send(line).is_ok(),
+            None => {
+                warn!("dropping command; no plugin connected");
+                false
+            }
+        }
+    }
+
+    pub async fn is_connected(&self) -> bool {
+        self.tx.lock().await.is_some()
+    }
+}
+
+/// Binds the loopback listener and accepts plugin connections forever. Each accepted connection
+/// runs until it drops, then we loop back to accept the next one. Events are forwarded to
+/// `event_tx`; the returned handle sends commands to whichever plugin is currently connected.
+///
+/// The **bound** address is returned alongside the handle rather than assumed to be the one asked
+/// for: `127.0.0.1:0` is a legitimate thing to configure (and what the tests use), and a log line
+/// echoing the request rather than the result is the kind that is wrong exactly when it matters.
+pub async fn serve(
+    addr: &str,
+    event_tx: mpsc::UnboundedSender<GameEvent>,
+) -> std::io::Result<(GameHandle, std::net::SocketAddr)> {
+    let listener = TcpListener::bind(addr).await?;
+    let bound = listener.local_addr()?;
+    info!(addr = %bound, "game link listening");
+
+    let handle = GameHandle::new();
+    let accept_handle = handle.clone();
+
+    tokio::spawn(async move {
+        loop {
+            match listener.accept().await {
+                Ok((stream, peer)) => {
+                    info!(%peer, "plugin connected");
+                    if let Err(e) = handle_connection(stream, &event_tx, &accept_handle).await {
+                        warn!(error = %e, "plugin connection ended");
+                    } else {
+                        info!("plugin disconnected");
+                    }
+                    accept_handle.set(None).await;
+                    // A disconnect is a fact the website should see without polling, so it rides
+                    // the same channel every other fact does. Nothing persists it: it is this
+                    // process's own observation, not something the game said — which is exactly
+                    // what `type: "control"` means (PROTOCOL.md §8.1). It is synthesised here
+                    // rather than anywhere else because this is the only place that knows.
+                    let _ = event_tx.send(GameEvent {
+                        kind: "link.down".to_string(),
+                        value: serde_json::json!({ "kind": "link.down", "type": "control" }),
+                    });
+                }
+                Err(e) => {
+                    warn!(error = %e, "accept failed");
+                    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+                }
+            }
+        }
+    });
+
+    Ok((handle, bound))
+}
+
+async fn handle_connection(
+    stream: tokio::net::TcpStream,
+    event_tx: &mpsc::UnboundedSender<GameEvent>,
+    handle: &GameHandle,
+) -> std::io::Result<()> {
+    stream.set_nodelay(true).ok();
+    let (read_half, mut write_half) = stream.into_split();
+
+    // Install the outbound command channel for this connection.
+    let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<String>();
+    handle.set(Some(cmd_tx)).await;
+
+    let mut reader = BufReader::new(read_half);
+    let mut lines = LineReader::default();
+
+    loop {
+        tokio::select! {
+            // Inbound: a line from the plugin.
+            result = lines.next(&mut reader) => {
+                match result? {
+                    Line::Eof => return Ok(()), // clean EOF: plugin closed
+                    Line::TooLong(bytes) => {
+                        // Deliberately not a disconnect. See MAX_INBOUND_LINE_BYTES.
+                        warn!(
+                            bytes,
+                            cap = MAX_INBOUND_LINE_BYTES,
+                            "inbound line over the cap; discarded"
+                        );
+                    }
+                    Line::Complete(line) => {
+                        let trimmed = line.trim_end();
+                        if !trimmed.is_empty() {
+                            match serde_json::from_str::<Value>(trimmed) {
+                                Ok(value) => {
+                                    let kind = value
+                                        .get("kind")
+                                        .and_then(|k| k.as_str())
+                                        .unwrap_or("")
+                                        .to_string();
+                                    let _ = event_tx.send(GameEvent { kind, value });
+                                }
+                                Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
+                            }
+                        }
+                    }
+                }
+            }
+            // Outbound: a command to write to the plugin.
+            cmd = cmd_rx.recv() => {
+                match cmd {
+                    Some(mut c) => {
+                        c.push('\n');
+                        write_half.write_all(c.as_bytes()).await?;
+                        write_half.flush().await?;
+                    }
+                    None => return Ok(()), // channel closed
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Drives `LineReader` over a byte slice, returning every outcome up to EOF.
+    async fn read_all(input: &[u8]) -> Vec<Line> {
+        let mut reader = BufReader::with_capacity(64, input);
+        let mut lines = LineReader::default();
+        let mut out = Vec::new();
+
+        loop {
+            match lines.next(&mut reader).await.unwrap() {
+                Line::Eof => break,
+                other => out.push(other),
+            }
+        }
+
+        out
+    }
+
+    fn complete(lines: &[Line]) -> Vec<&str> {
+        lines
+            .iter()
+            .filter_map(|l| match l {
+                Line::Complete(s) => Some(s.as_str()),
+                _ => None,
+            })
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn splits_on_newlines() {
+        let lines = read_all(b"{\"a\":1}\n{\"b\":2}\n").await;
+        assert_eq!(complete(&lines), vec!["{\"a\":1}", "{\"b\":2}"]);
+    }
+
+    /// The reader's buffer is 64 bytes here, so every one of these lines spans several `fill_buf`
+    /// chunks. Reassembly across chunks is the thing `read_line` would have done for us.
+    #[tokio::test]
+    async fn reassembles_across_chunks() {
+        let long = "x".repeat(500);
+        let input = format!("{}\n{}\n", long, long);
+        let lines = read_all(input.as_bytes()).await;
+
+        assert_eq!(complete(&lines), vec![long.as_str(), long.as_str()]);
+    }
+
+    /// The cap itself. The over-long line must be reported and thrown away, and — the part that
+    /// actually matters — the line *after* it must still arrive intact.
+    #[tokio::test]
+    async fn refuses_an_over_long_line_and_recovers() {
+        let mut input = Vec::new();
+        input.extend_from_slice(&b"a".repeat(MAX_INBOUND_LINE_BYTES + 10));
+        input.push(b'\n');
+        input.extend_from_slice(b"{\"kind\":\"pong\"}\n");
+
+        let lines = read_all(&input).await;
+
+        assert_eq!(lines.len(), 2);
+        assert!(
+            matches!(lines[0], Line::TooLong(n) if n >= MAX_INBOUND_LINE_BYTES),
+            "expected TooLong, got {:?}",
+            lines[0]
+        );
+        assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
+    }
+
+    /// A line of exactly the cap is legal; one byte more is not. Checking both sides is what says
+    /// the comparison is `>` rather than `>=`, which would silently cost a byte of the budget.
+    #[tokio::test]
+    async fn the_cap_is_inclusive() {
+        let at_cap = "b".repeat(MAX_INBOUND_LINE_BYTES);
+        let lines = read_all(format!("{}\n", at_cap).as_bytes()).await;
+        assert_eq!(complete(&lines).len(), 1);
+
+        let over = "b".repeat(MAX_INBOUND_LINE_BYTES + 1);
+        let lines = read_all(format!("{}\n", over).as_bytes()).await;
+        assert!(complete(&lines).is_empty());
+        assert!(matches!(lines[0], Line::TooLong(_)));
+    }
+
+    /// An over-long line whose terminator lands in the very chunk that crosses the cap: the reader
+    /// must not leave itself in `discarding` and eat the next line as well.
+    #[tokio::test]
+    async fn over_long_line_terminating_in_the_crossing_chunk() {
+        let mut input = Vec::new();
+        input.extend_from_slice(&b"c".repeat(MAX_INBOUND_LINE_BYTES + 1));
+        input.extend_from_slice(b"\n{\"kind\":\"pong\"}\n");
+
+        let lines = read_all(&input).await;
+
+        assert!(matches!(lines[0], Line::TooLong(_)));
+        assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
+    }
+
+    /// A partial line at EOF is dropped rather than delivered half-parsed. The plugin reconnects
+    /// and re-sends; half a JSON object is not something to hand to the event fan-out.
+    #[tokio::test]
+    async fn trailing_partial_line_at_eof_is_dropped() {
+        let lines = read_all(b"{\"a\":1}\n{\"b\":").await;
+        assert_eq!(complete(&lines), vec!["{\"a\":1}"]);
+    }
+
+    /// The end-to-end shape of the link, over a real socket: the plugin dials in, sends a hello,
+    /// and receives a command. This is the criterion phase 1 is judged on, minus the two peers.
+    #[tokio::test]
+    async fn a_dialling_plugin_is_accepted_and_can_be_commanded() {
+        use tokio::io::AsyncReadExt;
+
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let (handle, addr) = serve("127.0.0.1:0", tx).await.unwrap();
+
+        let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
+        client
+            .write_all(b"{\"kind\":\"server.hello\",\"serverId\":\"main\"}\n")
+            .await
+            .unwrap();
+
+        let ev = rx.recv().await.unwrap();
+        assert_eq!(ev.kind, "server.hello");
+        assert_eq!(ev.value["serverId"], "main");
+
+        assert!(handle.is_connected().await);
+        assert!(handle.send("{\"cmd\":\"ping\"}".to_string()).await);
+
+        let mut buf = [0u8; 64];
+        let n = client.read(&mut buf).await.unwrap();
+        assert_eq!(&buf[..n], b"{\"cmd\":\"ping\"}\n");
+    }
+}

sidecar/src/main.rs

@@ -0,0 +1,233 @@
+//! rust-link sidecar.
+//!
+//! Terminates the loopback link to a Rust game server's Oxide bridge plugin and exposes a
+//! website-facing HTTP surface: a WebSocket live feed and REST queries backed by SQLite.
+//!
+//! # Why the game does not listen
+//!
+//! The plugin is the TCP *client*; this process owns the listener. A Rust server therefore opens
+//! no extra port, and the only component the website can reach is this one. That invariant is
+//! inherited wholesale from the ServUO bridge — the footing changed (Oxide hooks instead of shard
+//! source) and the shape did not.
+//!
+//! # One server, one sidecar
+//!
+//! This binary serves exactly one game server. A community running six servers runs six pairs, and
+//! `module-rust` holds six clients; core never learns there is more than one. Nothing here is
+//! multiplexed, and nothing here should become multiplexed — the `serverId` on every frame exists
+//! so the *module* can tell its own clients apart, not so this process can.
+//!
+//! # Layout
+//!
+//! `main` does argument handling and nothing else; the sidecar proper lives in [`app`], which is
+//! parameterised on `ready`/`shutdown` so that a future service wrapper (the installer's phase)
+//! can supply the host's own start and stop without restructuring anything.
+
+mod app;
+mod cli;
+mod config;
+mod game;
+mod rpc;
+mod store;
+mod web;
+
+use tracing_subscriber::EnvFilter;
+
+/// Wire-protocol version between the website and this sidecar, and between this sidecar and the
+/// bridge plugin. Bump it whenever a frame's shape changes, so a mismatched peer is detected
+/// immediately (`409` on HTTP, a refusal in the log on the game link) rather than mis-parsed.
+///
+/// It is declared in **three** places and they must agree: here, in `Module-Rust`'s
+/// `module.json`, and in `Rust-Plugins`' `overlay.toml`. The installer refuses to pair a sidecar
+/// and an overlay that disagree, so a bump lands in the same change as the emitters it describes.
+///
+/// # Protocol 1 — the transport
+///
+/// Everything phase 1 defines, and deliberately nothing more:
+///
+///   * **`server.hello`** — the one event. The plugin sends it on every successful connect, not
+///     once at server start: this process restarts independently of the game, so anything the
+///     sidecar needs up front has to be re-sent per connection. It carries the `bootId`, which is
+///     how a game restart is told apart from a sidecar reconnect — the distinction the event
+///     system's `reconcile` will later hang off.
+///   * **`ping` / `pong`** — the heartbeat. The sidecar asks, the plugin answers. A `pong` is
+///     ephemeral chatter and is never persisted; it only moves `last_event`.
+///   * **`server.status`** — the one request/reply verb, correlated by `reqId`. It exists in
+///     phase 1 so the correlation path is exercised by something before anything depends on it.
+///
+/// Both directions are newline-delimited JSON over TCP. Outbound frames (plugin -> sidecar) carry
+/// `kind`; inbound frames (sidecar -> plugin) carry `cmd`.
+///
+/// # Protocol 2 — the read path
+///
+/// The catalogue: presence, deaths, chat, gathering, moderation and the wipe. Three things about
+/// it reach this file rather than only the plugin's:
+///
+///   * **Every frame carries `type`** — `event`, `snapshot`, `reply` or `control` — and this
+///     process files on THAT, never on `kind`. It is what keeps the sidecar a dumb forwarder while
+///     the catalogue grows: ten new event kinds are no change here at all.
+///   * **Every frame carries `serverId` and `wipeId`**, and both are lifted into indexed columns
+///     (the one migration shape the store's header predicted).
+///   * **`GET /feed`** is the ingest cursor, oldest-first, separate from `/events` so that no
+///     caller can get the other ordering by forgetting a parameter.
+///
+/// # Protocol 3 — identity
+///
+/// `POST /link/confirm`, the first route here that is not a GET, and the first message on this
+/// bridge the WEBSITE originates. It forwards a six-character code to the plugin and hands back
+/// what the plugin said. The codes live in the game's memory and nowhere else: putting the table
+/// here would give this process a credential and an opinion, and it is designed to have neither.
+///
+/// # Protocol 4 — the permission mirror
+///
+/// `GET /permissions/catalogue` and `POST /permissions/sync` (R2). The first command that WRITES
+/// to the game: the website sends the whole permission set it authors for this server and the
+/// plugin reconciles the store against it.
+///
+/// Nothing about that shape is visible in this process beyond two routes, and that is the
+/// dumb-forwarder property paying for itself a second time — protocol 4 adds the largest command
+/// on the bridge and touches neither the store nor the feed. The one thing this side owns is the
+/// envelope: `cmd` and `reqId` are written over whatever the caller sent, and a command that would
+/// not fit on the game link is refused here rather than discarded silently at the other end.
+///
+/// # Protocol 5 — configuration from the site
+///
+/// `GET /config/files`, `GET /config/file` and `POST /config/write` (R18). An admin edits a
+/// plugin's settings on the website; the plugin writes them, reloads whatever owns them, watches
+/// for the reload to announce itself, and **puts the old files back automatically** if it does
+/// not.
+///
+/// Two things about that reach this process. The write is the only route here that causes a write
+/// on the game host, and it is the only one whose reply routinely spends seconds rather than
+/// milliseconds — the plugin holds the correlation open across a reload and, at worst, across a
+/// rollback as well. `web::CONFIG_RELOAD_WINDOW` is that budget, mirrored from the plugin, and a
+/// test asserts the pairing rather than trusting it.
+///
+/// # Protocol 6 — first-party clans
+///
+/// One board (`clans`) and five events (`clan.created`, `clan.disbanded`, `clan.member.added`,
+/// `clan.member.left`, `clan.member.kicked`), from which the website builds core's Teams (R5).
+///
+/// **Nothing in this process changed except this number**, and that is protocol 2's promise
+/// kept: the board is filed by `type: "snapshot"` and the events by `type: "event"`, so the
+/// sidecar needs no arm for any of them. The bump exists for the other two declaration sites —
+/// a website that reads clans must not pair with a plugin that never sends them.
+///
+/// One property of this process does bear on the board: [`game::MAX_INBOUND_LINE_BYTES`]
+/// discards a line over 1 MiB outright. The plugin bounds the board well inside it and says
+/// `truncated` when it had to stop, because a board that never arrived would read as a server
+/// with no clans.
+///
+/// # Protocol 7 — a raid frame that names who lives there
+///
+/// `entity.destroyed` widens to doors, external walls and the tool cupboard, and gains the
+/// cupboard's `buildingId` and its `authorized` list, which is what lets the website send the
+/// raid alert to the people whose base it was (the module's PLAN.md §25).
+///
+/// **Again nothing here changed but this number.** The frame is an `event`, stored and served as
+/// it arrived. The bump exists because a website that alerts on `authorized` must not pair with a
+/// plugin that never sends it — against protocol 6 it would read every raid as a base with no
+/// cupboard, and alert nobody while looking healthy.
+///
+/// # Protocol 8 — the leases
+///
+/// `GET /lease`, `POST /lease` and `POST /lease/release`: an event borrowing a value and giving
+/// it back (the module's PLAN.md §27). Three correlated round trips and one event,
+/// `lease.expired`, which is filed like every other event. The allowlist, the bounds, the
+/// seven-day ceiling and the deadline timer all live in the plugin; the ledger lives on the
+/// website. This process moves lines between them and learns nothing about either, which is why
+/// the whole protocol is three thin forwards here.
+///
+/// # Protocol 9 — the world verbs
+///
+/// `GET /world/monuments`, `GET /world/owned`, `POST /world/zone`, `POST /world/place` and
+/// `POST /world/revert`: what an event makes in the world — a zone, crates, NPCs — and gives back
+/// (the module's PLAN.md §28). Five more thin forwards. The allowlist, the bounds, the monument
+/// vocabulary and the registry of what each run owns all live in the plugin.
+///
+/// # Protocol 10 — the rewards
+///
+/// `POST /tally/open`, `GET /tally/snapshot`, `POST /tally/close`, `GET /kits` and `POST /chat`:
+/// who took part in a run, the kits a reward can name, and one line in the server's chat (the
+/// module's PLAN.md §29). Five more thin forwards. `perm.sync` also gains a `credits` field, which
+/// passes through untouched like the rest of that body. The tally, the kit catalogue and the chat
+/// memory all live in the plugin.
+///
+/// `docs/rust-link/PROTOCOL.md` is the specification — §8 the read path, §9 identity, §10 the
+/// mirror, §11 configuration, §12 clans, §13 the raid frame, §14 the leases, §15 the world verbs,
+/// §16 the rewards; this constant is one of its four declaration sites.
+pub const PROTOCOL_VERSION: u32 = 10;
+
+fn main() -> anyhow::Result<()> {
+    let args = match cli::parse(std::env::args().skip(1)) {
+        Ok(args) => args,
+        Err(msg) => {
+            eprintln!("rust-link-sidecar: {msg}\n\n{}", cli::USAGE);
+            std::process::exit(2);
+        }
+    };
+
+    match args.mode {
+        cli::Mode::Help => {
+            print!("{}", cli::USAGE);
+            return Ok(());
+        }
+        cli::Mode::Version => {
+            println!(
+                "rust-link-sidecar {} (protocol {})",
+                env!("CARGO_PKG_VERSION"),
+                PROTOCOL_VERSION
+            );
+            return Ok(());
+        }
+        // Tracing stays uninitialized here on purpose: the subscriber writes to stdout, and on this
+        // path stdout is the document. An installer parses it.
+        cli::Mode::PrintConfig => {
+            let loaded = config::Config::load(args.config.as_deref())?;
+            println!("{:#}", config::describe(&loaded));
+            return Ok(());
+        }
+        cli::Mode::Run => {}
+    }
+
+    init_console_tracing();
+
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()?;
+
+    runtime.block_on(app::run(args.config.as_deref(), || {}, shutdown_signal()))
+}
+
+/// Logging for a foreground run: human-readable, on stdout.
+pub fn init_console_tracing() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
+        )
+        .init();
+}
+
+/// Resolves on Ctrl-C, and on `SIGTERM` where there is one.
+async fn shutdown_signal() {
+    #[cfg(unix)]
+    {
+        use tokio::signal::unix::{signal, SignalKind};
+        let mut term = match signal(SignalKind::terminate()) {
+            Ok(s) => s,
+            Err(_) => {
+                let _ = tokio::signal::ctrl_c().await;
+                return;
+            }
+        };
+        tokio::select! {
+            _ = tokio::signal::ctrl_c() => {}
+            _ = term.recv() => {}
+        }
+    }
+
+    #[cfg(not(unix))]
+    {
+        let _ = tokio::signal::ctrl_c().await;
+    }
+}

sidecar/src/rpc.rs

@@ -0,0 +1,135 @@
+//! Request/reply correlation over the one game socket.
+//!
+//! REST is synchronous ("what is the server doing right now"); the game link is an async stream of
+//! lines. This bridges them: a call registers a pending entry under a correlation id, sends the
+//! command, and awaits a reply carrying that id. The event loop routes any incoming line whose
+//! correlation id is pending back to the waiting caller; everything else flows on as a normal
+//! event.
+//!
+//! One correlation field is recognised in protocol 1 — **`reqId`** — and it is a process-unique
+//! counter. The UO bridge grew two more over eight protocol versions because callers there supply
+//! their own ids for some verbs; that is a reason to keep the routing table keyed by string, not a
+//! reason to invent the extra keys now.
+//!
+//! **The timeout here is a ceiling every later command budget sits under.** Core classifies a
+//! budget overrun as retryable unconditionally — it cannot ask the game, because the action is
+//! still awaiting a socket — so an action whose `budgetMs` exceeds [`REPLY_TIMEOUT`] can never
+//! report `retry: false`. Derive one from the other rather than writing both down.
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use serde_json::Value;
+use tokio::sync::{oneshot, Mutex};
+
+use crate::game::GameHandle;
+
+/// How long a correlated call waits for its reply before giving up.
+pub const REPLY_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[derive(Clone)]
+pub struct Rpc {
+    pending: Arc<Mutex<HashMap<String, oneshot::Sender<Value>>>>,
+    counter: Arc<AtomicU64>,
+}
+
+#[derive(Debug)]
+pub enum RpcError {
+    NoPlugin,
+    Timeout,
+}
+
+impl Rpc {
+    pub fn new() -> Self {
+        Self {
+            pending: Arc::new(Mutex::new(HashMap::new())),
+            counter: Arc::new(AtomicU64::new(1)),
+        }
+    }
+
+    pub fn next_req_id(&self) -> String {
+        format!("r-{}", self.counter.fetch_add(1, Ordering::Relaxed))
+    }
+
+    /// Sends `command` to the plugin and awaits the reply correlated by `corr_val`. The command
+    /// must already contain the correlation field (`reqId`) set to `corr_val`.
+    pub async fn call(
+        &self,
+        game: &GameHandle,
+        command: Value,
+        corr_val: &str,
+    ) -> Result<Value, RpcError> {
+        let (tx, rx) = oneshot::channel();
+
+        self.pending.lock().await.insert(corr_val.to_string(), tx);
+
+        if !game.send(command.to_string()).await {
+            self.pending.lock().await.remove(corr_val);
+            return Err(RpcError::NoPlugin);
+        }
+
+        match tokio::time::timeout(REPLY_TIMEOUT, rx).await {
+            Ok(Ok(value)) => Ok(value),
+            _ => {
+                // Both the timeout and a dropped sender land here, and both must clear the entry:
+                // a pending map that only ever grows is a leak keyed by a counter.
+                self.pending.lock().await.remove(corr_val);
+                Err(RpcError::Timeout)
+            }
+        }
+    }
+
+    /// If this incoming value correlates to a pending call, complete it and return true (the value
+    /// was a reply, not a broadcast event). Otherwise return false.
+    pub async fn try_route(&self, value: &Value) -> bool {
+        let corr = match value.get("reqId").and_then(|v| v.as_str()) {
+            Some(c) => c.to_string(),
+            None => return false,
+        };
+
+        let sender = self.pending.lock().await.remove(&corr);
+        match sender {
+            Some(tx) => {
+                let _ = tx.send(value.clone());
+                true
+            }
+            None => false,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[tokio::test]
+    async fn an_unknown_req_id_is_not_a_reply() {
+        let rpc = Rpc::new();
+        assert!(!rpc.try_route(&json!({"kind": "server.hello"})).await);
+        assert!(!rpc.try_route(&json!({"reqId": "r-999"})).await);
+    }
+
+    #[tokio::test]
+    async fn req_ids_are_unique_within_a_process() {
+        let rpc = Rpc::new();
+        let a = rpc.next_req_id();
+        let b = rpc.next_req_id();
+        assert_ne!(a, b);
+    }
+
+    /// A reply routed while nothing is waiting must flow on as an ordinary event rather than be
+    /// swallowed — that is the difference between a late reply and a lost one.
+    #[tokio::test]
+    async fn a_late_reply_falls_through_to_the_event_path() {
+        let rpc = Rpc::new();
+        let (tx, _rx) = oneshot::channel();
+        rpc.pending.lock().await.insert("r-1".into(), tx);
+
+        assert!(rpc.try_route(&json!({"reqId": "r-1"})).await);
+        // Second delivery of the same id: the entry is gone, so it is an event now.
+        assert!(!rpc.try_route(&json!({"reqId": "r-1"})).await);
+    }
+}

sidecar/src/store.rs

@@ -0,0 +1,671 @@
+//! SQLite persistence: the event history, and the boards holding what is true right now.
+//!
+//! 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 the columns it must *index* lifted out. 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.
+//!
+//! Protocol 2 is the first version that needed one — `server_id` and `wipe_id` (PROTOCOL.md §8.9)
+//! — and it is applied the way this project applies every schema change: as an `ALTER` guarded by a
+//! column check, never as an edit to the `CREATE`, because `CREATE TABLE IF NOT EXISTS` does
+//! nothing at all against a database that already has the table and an edited column would reach
+//! fresh installs only.
+
+use std::path::Path;
+
+use serde_json::{json, Value};
+use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
+use sqlx::{Row, SqlitePool};
+use tracing::{info, warn};
+
+const SCHEMA: &str = "
+CREATE TABLE IF NOT EXISTS events (
+    id        INTEGER PRIMARY KEY AUTOINCREMENT,
+    t         INTEGER NOT NULL,
+    kind      TEXT    NOT NULL,
+    server_id TEXT,
+    wipe_id   TEXT,
+    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);
+
+-- Boards: current state, one row per kind, replaced whole. A board in chapter 4's sense — state
+-- with exactly one producer, re-sent on every connect — rather than a history. Protocol 1 had one
+-- of these hard-coded as `server_state`; protocol 2 has two and will have more, so the kind is a
+-- key rather than a table name.
+CREATE TABLE IF NOT EXISTS boards (
+    kind    TEXT    PRIMARY KEY,
+    t       INTEGER NOT NULL,
+    json    TEXT    NOT NULL
+);
+
+-- Protocol 1's single board. Kept so that a sidecar upgraded in place can carry its contents over
+-- (see `migrate`); nothing writes to it any more.
+CREATE TABLE IF NOT EXISTS server_state (
+    id      INTEGER PRIMARY KEY CHECK (id = 1),
+    t       INTEGER NOT NULL,
+    json    TEXT    NOT NULL
+);
+";
+
+/// The board holding the last `server.hello`. Named once here rather than spelled at four call
+/// sites, because it is the one board with a REST route of its own.
+pub const SERVER_BOARD: &str = "server.hello";
+
+/// One row of the ingest feed: a stored event, with the identity a cursor needs.
+#[derive(Debug, Clone)]
+pub struct FeedItem {
+    pub id: i64,
+    pub t: i64,
+    pub kind: String,
+    pub frame: Value,
+}
+
+impl FeedItem {
+    pub fn to_json(&self) -> Value {
+        json!({ "id": self.id, "t": self.t, "kind": self.kind, "frame": self.frame })
+    }
+}
+
+#[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?;
+
+        let store = Self { pool };
+        store.migrate().await?;
+
+        info!(%path, "store ready");
+        Ok(store)
+    }
+
+    /// Brings a database created by an older protocol up to this one.
+    ///
+    /// Two steps, both idempotent, both safe to run on a fresh database where they do nothing:
+    /// add the columns protocol 2 indexes on, and carry protocol 1's single board into `boards`.
+    ///
+    /// The board carry-over matters more than it looks: without it, an upgraded sidecar answers
+    /// `204` for `/server` until the game next connects, and the website reads that as *this
+    /// server has never been heard from* — the site loses a server it has been rendering for
+    /// weeks, at the exact moment somebody upgraded the bridge.
+    async fn migrate(&self) -> anyhow::Result<()> {
+        for (column, ddl) in [
+            ("server_id", "ALTER TABLE events ADD COLUMN server_id TEXT"),
+            ("wipe_id", "ALTER TABLE events ADD COLUMN wipe_id TEXT"),
+        ] {
+            if !self.has_column("events", column).await? {
+                sqlx::query(ddl).execute(&self.pool).await?;
+                info!(column, "events: column added");
+            }
+        }
+
+        // Indexed after the columns exist, and in the same idempotent spirit.
+        sqlx::query("CREATE INDEX IF NOT EXISTS idx_events_wipe_id ON events (wipe_id, id DESC)")
+            .execute(&self.pool)
+            .await?;
+
+        let carried: Option<(i64, String)> = sqlx::query_as(
+            "SELECT t, json FROM server_state WHERE id = 1
+             AND NOT EXISTS (SELECT 1 FROM boards WHERE kind = ?)",
+        )
+        .bind(SERVER_BOARD)
+        .fetch_optional(&self.pool)
+        .await?;
+
+        if let Some((t, json)) = carried {
+            self.put_board(SERVER_BOARD, t, &json).await?;
+            info!("carried the protocol 1 server board into boards");
+        }
+
+        Ok(())
+    }
+
+    async fn has_column(&self, table: &str, column: &str) -> anyhow::Result<bool> {
+        // `PRAGMA table_info` does not take a bind parameter for the table name, which is why this
+        // is formatted. Both call sites pass a literal; nothing here is reachable from a request.
+        let rows = sqlx::query(&format!("PRAGMA table_info({table})"))
+            .fetch_all(&self.pool)
+            .await?;
+
+        Ok(rows
+            .iter()
+            .any(|r| r.get::<String, _>("name").eq_ignore_ascii_case(column)))
+    }
+
+    /// 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.
+    ///
+    /// `server_id` and `wipe_id` are lifted out of the frame by the caller and stored as columns as
+    /// well as remaining in the JSON. Duplicated deliberately: the column is what an index and a
+    /// `WHERE` can reach, and the JSON is what stays correct when the columns change.
+    pub async fn insert_event(
+        &self,
+        t: i64,
+        kind: &str,
+        server_id: Option<&str>,
+        wipe_id: Option<&str>,
+        json: &str,
+    ) -> anyhow::Result<()> {
+        sqlx::query(
+            "INSERT INTO events (t, kind, server_id, wipe_id, json) VALUES (?, ?, ?, ?, ?)",
+        )
+        .bind(t)
+        .bind(kind)
+        .bind(server_id)
+        .bind(wipe_id)
+        .bind(json)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    /// Most-recent events, **newest first**, optionally filtered by kind and by wipe.
+    ///
+    /// For a human, an admin screen, or a point-in-time look. A consumer that must not miss a row
+    /// wants [`Store::feed`] instead — see its documentation for why these are two functions and
+    /// not one with a flag.
+    pub async fn recent(
+        &self,
+        kind: Option<&str>,
+        wipe: Option<&str>,
+        limit: i64,
+    ) -> anyhow::Result<Vec<Value>> {
+        let limit = limit.clamp(1, 1000);
+
+        // Built rather than branched four ways: two optional filters is four combinations, and the
+        // fourth is always the one nobody tested. The bindings stay parameterised.
+        let mut sql = String::from("SELECT json FROM events WHERE 1 = 1");
+
+        if kind.is_some() {
+            sql.push_str(" AND kind = ?");
+        }
+        if wipe.is_some() {
+            sql.push_str(" AND wipe_id = ?");
+        }
+
+        sql.push_str(" ORDER BY id DESC LIMIT ?");
+
+        let mut query = sqlx::query(&sql);
+
+        if let Some(k) = kind {
+            query = query.bind(k);
+        }
+        if let Some(w) = wipe {
+            query = query.bind(w);
+        }
+
+        let rows = query.bind(limit).fetch_all(&self.pool).await?;
+        Ok(parse_json_column(rows))
+    }
+
+    /// The ingest cursor: events **after** `since`, **oldest first**.
+    ///
+    /// This is a separate function from [`Store::recent`], and the route on top of it is a separate
+    /// route, for one reason: a single route whose ordering depends on a query parameter serves the
+    /// other ordering to every caller that forgets it, and for the ingesting caller that means
+    /// advancing its cursor past rows it never read. Silently, and once per deployment mistake.
+    ///
+    /// Returns the page and whether it filled — a consumer an hour behind drains at its own pace
+    /// rather than guessing from a count.
+    pub async fn feed(&self, since: i64, limit: i64) -> anyhow::Result<(Vec<FeedItem>, bool)> {
+        let limit = limit.clamp(1, 1000);
+
+        let rows = sqlx::query(
+            "SELECT id, t, kind, json FROM events WHERE id > ? ORDER BY id ASC LIMIT ?",
+        )
+        .bind(since)
+        .bind(limit)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let more = rows.len() as i64 == limit;
+
+        let items = rows
+            .into_iter()
+            .filter_map(|r| {
+                let json: String = r.get("json");
+                serde_json::from_str(&json).ok().map(|frame| FeedItem {
+                    id: r.get("id"),
+                    t: r.get("t"),
+                    kind: r.get("kind"),
+                    frame,
+                })
+            })
+            .collect();
+
+        Ok((items, more))
+    }
+
+    /// The highest event id in the store, or 0 when it is empty.
+    ///
+    /// A consumer starting from nothing uses this to begin at the *end* rather than replaying the
+    /// whole history it has no use for — a fresh module against a sidecar that has been running for
+    /// a month wants what happens next, not a fortnight of deaths.
+    pub async fn last_event_id(&self) -> anyhow::Result<i64> {
+        let row = sqlx::query("SELECT COALESCE(MAX(id), 0) AS id FROM events")
+            .fetch_one(&self.pool)
+            .await?;
+        Ok(row.get("id"))
+    }
+
+    /// Replaces one board. Called for every snapshot frame, which the plugin re-sends on every
+    /// connect and on a cadence — so this is an upsert by construction, not by accident.
+    pub async fn put_board(&self, kind: &str, t: i64, json: &str) -> anyhow::Result<()> {
+        sqlx::query(
+            "INSERT INTO boards (kind, t, json) VALUES (?, ?, ?)
+             ON CONFLICT(kind) DO UPDATE SET t = excluded.t, json = excluded.json",
+        )
+        .bind(kind)
+        .bind(t)
+        .bind(json)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
+    /// One board, or `None` if the game has never sent it.
+    ///
+    /// 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 board(&self, kind: &str) -> anyhow::Result<Option<Value>> {
+        let row = sqlx::query("SELECT json FROM boards WHERE kind = ?")
+            .bind(kind)
+            .fetch_optional(&self.pool)
+            .await?;
+        Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
+    }
+
+    /// Every board, keyed by kind. What a consumer reads once on connect to know the present
+    /// before it starts following the story.
+    pub async fn all_boards(&self) -> anyhow::Result<serde_json::Map<String, Value>> {
+        let rows = sqlx::query("SELECT kind, json FROM boards ORDER BY kind")
+            .fetch_all(&self.pool)
+            .await?;
+
+        let mut out = serde_json::Map::new();
+
+        for row in rows {
+            let kind: String = row.get("kind");
+            if let Ok(v) = serde_json::from_str::<Value>(&row.get::<String, _>("json")) {
+                out.insert(kind, v);
+            }
+        }
+
+        Ok(out)
+    }
+
+    /// Deletes events older than `retain_days`, returning how many went.
+    ///
+    /// **Boards are never pruned**, and that asymmetry is the design rather than an oversight: a
+    /// board is one row per kind holding what is true now, and deleting it would make a server the
+    /// site has rendered for weeks look like one that has never connected. History is bounded
+    /// because it grows; the present is not, because it does not.
+    ///
+    /// Safe to be aggressive here because the *permanent* record lives on the website — per-wipe
+    /// rollups in the module's own tables (R12) — and this database sits on a game host whose disk
+    /// belongs to the operator.
+    pub async fn prune(&self, retain_days: i64) -> anyhow::Result<u64> {
+        if retain_days <= 0 {
+            return Ok(0); // retention off; an operator who wants everything keeps everything
+        }
+
+        let cutoff = now_ms() - retain_days * 86_400_000;
+
+        let done = sqlx::query("DELETE FROM events WHERE t < ?")
+            .bind(cutoff)
+            .execute(&self.pool)
+            .await?;
+
+        let n = done.rows_affected();
+
+        if n > 0 {
+            info!(pruned = n, retain_days, "pruned old events");
+        }
+
+        Ok(n)
+    }
+}
+
+fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+/// 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()
+}
+
+/// Warns once about a store write that failed. Persistence failures must never stop the live feed,
+/// so every caller logs and carries on; this keeps them saying the same thing.
+pub fn warn_write(what: &str, e: &anyhow::Error) {
+    warn!(error = %e, "{what}");
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    async fn store() -> Store {
+        Store::open(":memory:").await.unwrap()
+    }
+
+    async fn insert(s: &Store, t: i64, kind: &str, wipe: Option<&str>) {
+        s.insert_event(
+            t,
+            kind,
+            Some("main"),
+            wipe,
+            &json!({"kind": kind, "t": t}).to_string(),
+        )
+        .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. Every table 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);
+
+        insert(&s, 1, "k", None).await;
+        s.put_board(SERVER_BOARD, 1, "{}").await.unwrap();
+        assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 1);
+        assert!(s.board(SERVER_BOARD).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;
+        insert(&s, 1, "server.hello", None).await;
+        insert(&s, 2, "other", None).await;
+        insert(&s, 3, "server.hello", None).await;
+
+        let all = s.recent(None, None, 10).await.unwrap();
+        assert_eq!(all.len(), 3);
+        assert_eq!(all[0]["t"], 3);
+
+        let hellos = s.recent(Some("server.hello"), None, 10).await.unwrap();
+        assert_eq!(hellos.len(), 2);
+        assert_eq!(hellos[0]["t"], 3);
+    }
+
+    /// R12 in one test: a wipe splits the history without erasing any of it.
+    #[tokio::test]
+    async fn events_filter_by_wipe_without_losing_the_other_wipe() {
+        let s = store().await;
+        insert(&s, 1, "player.death", Some("w-a")).await;
+        insert(&s, 2, "player.death", Some("w-a")).await;
+        insert(&s, 3, "player.death", Some("w-b")).await;
+
+        assert_eq!(s.recent(None, Some("w-a"), 10).await.unwrap().len(), 2);
+        assert_eq!(s.recent(None, Some("w-b"), 10).await.unwrap().len(), 1);
+        assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 3);
+        // Both filters at once is the combination that is easy to build wrong.
+        assert_eq!(
+            s.recent(Some("player.death"), Some("w-b"), 10)
+                .await
+                .unwrap()
+                .len(),
+            1
+        );
+    }
+
+    /// The cursor's two properties, and they are the ones a consumer's correctness rests on:
+    /// oldest first, and strictly after the id it was given.
+    #[tokio::test]
+    async fn the_feed_is_a_cursor_and_runs_oldest_first() {
+        let s = store().await;
+        for i in 1..=5 {
+            insert(&s, i, "player.death", None).await;
+        }
+
+        let (page, more) = s.feed(0, 2).await.unwrap();
+        assert_eq!(page.len(), 2);
+        assert!(more, "a full page must say there is more");
+        assert_eq!(page[0].t, 1);
+        assert_eq!(page[1].t, 2);
+
+        let (page, more) = s.feed(page[1].id, 10).await.unwrap();
+        assert_eq!(page.len(), 3);
+        assert!(!more, "a short page is the end of the queue");
+        assert_eq!(page[0].t, 3);
+
+        // The cursor is exclusive; re-reading from the last id delivers nothing twice.
+        let (page, _) = s.feed(page[2].id, 10).await.unwrap();
+        assert!(page.is_empty());
+    }
+
+    #[tokio::test]
+    async fn a_fresh_consumer_can_start_at_the_end() {
+        let s = store().await;
+        assert_eq!(s.last_event_id().await.unwrap(), 0);
+
+        for i in 1..=3 {
+            insert(&s, i, "k", None).await;
+        }
+
+        let last = s.last_event_id().await.unwrap();
+        assert_eq!(last, 3);
+        assert!(s.feed(last, 10).await.unwrap().0.is_empty());
+    }
+
+    /// 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 a_board_is_absent_until_a_snapshot_arrives() {
+        let s = store().await;
+        assert!(s.board(SERVER_BOARD).await.unwrap().is_none());
+
+        s.put_board(SERVER_BOARD, 1, &json!({"serverId": "main"}).to_string())
+            .await
+            .unwrap();
+        assert_eq!(
+            s.board(SERVER_BOARD).await.unwrap().unwrap()["serverId"],
+            "main"
+        );
+    }
+
+    /// A board holds exactly one row however many times the plugin reconnects, and the boards are
+    /// independent of one another.
+    #[tokio::test]
+    async fn a_second_snapshot_replaces_the_first_of_its_own_kind_only() {
+        let s = store().await;
+        s.put_board(SERVER_BOARD, 1, &json!({"bootId": "a"}).to_string())
+            .await
+            .unwrap();
+        s.put_board(SERVER_BOARD, 2, &json!({"bootId": "b"}).to_string())
+            .await
+            .unwrap();
+        s.put_board("players.online", 2, &json!({"count": 4}).to_string())
+            .await
+            .unwrap();
+
+        assert_eq!(s.board(SERVER_BOARD).await.unwrap().unwrap()["bootId"], "b");
+        assert_eq!(
+            s.board("players.online").await.unwrap().unwrap()["count"],
+            4
+        );
+
+        let all = s.all_boards().await.unwrap();
+        assert_eq!(all.len(), 2);
+    }
+
+    #[tokio::test]
+    async fn the_limit_is_clamped_rather_than_trusted() {
+        let s = store().await;
+        for i in 0..5 {
+            insert(&s, i, "k", None).await;
+        }
+        // 0 and negatives would otherwise mean "no rows" and "SQLite's unlimited" respectively.
+        assert_eq!(s.recent(None, None, 0).await.unwrap().len(), 1);
+        assert_eq!(s.recent(None, None, -1).await.unwrap().len(), 1);
+        assert_eq!(s.recent(None, None, 100_000).await.unwrap().len(), 5);
+        assert_eq!(s.feed(0, 0).await.unwrap().0.len(), 1);
+    }
+
+    /// Retention deletes history and leaves the present alone. The second half is the half worth
+    /// asserting: a pruned board is a server that has "never connected".
+    #[tokio::test]
+    async fn pruning_bounds_the_history_and_never_touches_a_board() {
+        let s = store().await;
+        let old = now_ms() - 30 * 86_400_000;
+
+        insert(&s, old, "player.death", None).await;
+        insert(&s, now_ms(), "player.death", None).await;
+        s.put_board(SERVER_BOARD, old, &json!({"serverId": "main"}).to_string())
+            .await
+            .unwrap();
+
+        assert_eq!(s.prune(14).await.unwrap(), 1);
+        assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 1);
+        assert!(s.board(SERVER_BOARD).await.unwrap().is_some());
+
+        // Retention off keeps everything, which is a supported configuration rather than a bug.
+        insert(&s, old, "player.death", None).await;
+        assert_eq!(s.prune(0).await.unwrap(), 0);
+        assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 2);
+    }
+
+    /// The upgrade path, on a real file because that is the only place it can happen: a protocol 1
+    /// database has `server_state` and no `wipe_id`, and opening it with this build must produce a
+    /// store that still knows which server it is holding.
+    #[tokio::test]
+    async fn a_protocol_1_database_is_migrated_in_place() {
+        let dir = std::env::temp_dir().join(format!("rust-link-migrate-{}", std::process::id()));
+        let path = dir.join("old.db");
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+
+        // Exactly protocol 1's schema, written by hand so the test does not depend on this file
+        // still being able to produce it.
+        let opts = SqliteConnectOptions::new()
+            .filename(&path)
+            .create_if_missing(true);
+        let pool = SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect_with(opts)
+            .await
+            .unwrap();
+        sqlx::query(
+            "CREATE TABLE events (id INTEGER PRIMARY KEY AUTOINCREMENT, t INTEGER NOT NULL,
+                                  kind TEXT NOT NULL, json TEXT NOT NULL);
+             CREATE TABLE server_state (id INTEGER PRIMARY KEY CHECK (id = 1), t INTEGER NOT NULL,
+                                  json TEXT NOT NULL);",
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+        sqlx::query("INSERT INTO events (t, kind, json) VALUES (1, 'server.hello', '{\"t\":1}')")
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query(
+            "INSERT INTO server_state (id, t, json) VALUES (1, 1, '{\"serverId\":\"legacy\"}')",
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+        pool.close().await;
+
+        let s = Store::open(path.to_str().unwrap()).await.unwrap();
+
+        // The columns arrived, the old rows survived with them empty, and the board came across —
+        // so an upgraded sidecar does not report a server it has been serving for weeks as one it
+        // has never heard of.
+        assert!(s.has_column("events", "wipe_id").await.unwrap());
+        assert_eq!(s.recent(None, None, 10).await.unwrap().len(), 1);
+        assert_eq!(
+            s.board(SERVER_BOARD).await.unwrap().unwrap()["serverId"],
+            "legacy"
+        );
+
+        // And it is idempotent: opening again must not fail on an ALTER that already ran.
+        drop(s);
+        let again = Store::open(path.to_str().unwrap()).await.unwrap();
+        assert!(again.board(SERVER_BOARD).await.unwrap().is_some());
+
+        drop(again);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}