Sidecar: REST query layer

rpc.rs bridges synchronous REST to the async shard stream. A call registers a
pending entry under a correlation id, sends the command, and awaits the reply
(10s timeout). The event loop routes any incoming line whose id is pending back
to the waiting caller; everything else stays a live event and is broadcast. Three
correlation fields are recognized, matching what the plugin echoes: reqId
(queries), code (link.confirm), id (towncrier).

web.rs adds the routes: GET /char/{account}/{slot}, /char/serial/{serial},
/roster/{account}, /vendors/{account}; POST /link/confirm, POST /towncrier,
DELETE /towncrier/{id}. A shard *.error reply maps to 404 or 400; no shard -> 503;
no reply in time -> 504.

Verified end to end against the live shard: roster and full char profile returned
as JSON (reqId correlation visible as r-1, r-2, ...), an unknown account returned
bridge.error as HTTP 404, vendor snapshot returned seed_000's two shops, towncrier
publish and remove returned towncrier.ok, and a bad link code returned link.error
as 404. The website can now query the game and push commands, all correlated over
the single loopback socket, all through the sidecar the game never directly
exposes.

Only SQLite persistence remains on the sidecar.

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

sidecar/README.md

@@ -24,16 +24,33 @@ Binds `127.0.0.1:7788` and waits for the shard to connect. Boot the shard (or it
 |-------|-------|
 | Shard link (`shard.rs`) | **done** — accepts the shard, reads events, sends commands, re-accepts on disconnect. Verified against the live shard: received `server.hello`, round-tripped a `ping`→`pong`, and reconnected after a sidecar restart. |
 | WebSocket feed (`web.rs`) | **done** — `/ws` fans every shard event out to connected clients via a `broadcast`. Verified: a WS client received `ws.hello` then live `pong` events relayed from the shard. Live-only, no replay. |
-| REST queries (char profile, roster, vendor snapshot, link submit) | not started |
+| REST queries (`rpc.rs` + `web.rs`) | **done** — synchronous queries and commands, correlated to shard replies by id. Verified end-to-end against the live shard, success and error paths. |
 | SQLite persistence (event history, economy, cached profiles, link map) | not started |
 
-The web server binds `127.0.0.1:8080` by default (`WEB_ADDR` in `main.rs`). Routes: `GET /health` → `ok`, `GET /ws` → the live feed. Widen the bind and add auth before exposing it off-host.
+The web server binds `127.0.0.1:8080` by default (`WEB_ADDR` in `main.rs`). Widen the bind and add auth before exposing it off-host.
+
+### Routes
+
+| Method | Path | Shard command | Reply |
+|--------|------|---------------|-------|
+| GET | `/health` | — | `ok` |
+| GET | `/ws` | — | live event feed (WebSocket) |
+| GET | `/char/{account}/{slot}` | `char.request` | `char.profile` |
+| GET | `/char/serial/{serial}` | `char.request` | `char.profile` |
+| GET | `/roster/{account}` | `account.roster` | `account.roster` |
+| GET | `/vendors/{account}` | `vendor.snapshot` | `vendor.snapshot` |
+| POST | `/link/confirm` `{code, websiteUserId}` | `link.confirm` | `link.ok` / `link.error` |
+| POST | `/towncrier` `{id, lines, durationSec}` | `towncrier.add` | `towncrier.ok` / `towncrier.error` |
+| DELETE | `/towncrier/{id}` | `towncrier.remove` | `towncrier.ok` / `towncrier.error` |
+
+A shard `*.error` reply maps to HTTP 404 (unknown/not-found) or 400 (bad request). No shard connected → 503; no reply within 10 s → 504.
 
 ## Design
 
 - **`shard.rs`** — `serve()` binds the listener and accepts shard connections in a loop. Each connection splits into read/write halves: the read half parses newline-JSON into `ShardEvent { kind, value }` and forwards them; the write half drains an mpsc of command lines. `ShardHandle::send` posts a command to whichever shard is currently connected, and **drops with a warning if none is** — a website query during a shard outage should fail fast and retry, not queue behind a reconnect. Live *events* that must survive an outage are buffered by the shard, not here.
 - **`web.rs`** — the website-facing HTTP surface (axum). `AppState` holds the `broadcast::Sender<String>`; each `/ws` client subscribes and forwards every event as a text frame. A client that lags past the broadcast buffer is warned and kept live (it just misses events) rather than stalling the others. This side *may* be exposed beyond loopback — it is the gatekeeper, so add auth when you do.
-- **`main.rs`** — wires it together: the shard event loop logs each event and `broadcast::send`s it to the WS feed. Later phases also persist to SQLite here and turn REST calls into shard commands via `ShardHandle`.
+- **`rpc.rs`** — request/reply correlation over the one shard socket. A REST call registers a pending entry under a correlation id, sends the command, and awaits the reply (10 s timeout). The event loop routes any incoming line whose id is pending back to the waiter; everything else flows on as a live event. Recognizes three correlation fields, matching what the plugin echoes: `reqId` (queries), `code` (link), `id` (town-crier).
+- **`main.rs`** — wires it together: the shard event loop first tries to route each line as an RPC reply; if it isn't one, the line is a live event, logged and broadcast to WS. Later phases persist to SQLite here.
 
 ## Wire protocol
 

sidecar/src/main.rs

@@ -3,6 +3,7 @@
 //! Terminates the loopback link to the ServUO shard and exposes a website-facing HTTP surface. So
 //! far: the shard link (bidirectional) and a WebSocket live feed. REST queries and SQLite come next.
 
+mod rpc;
 mod shard;
 mod web;
 
@@ -25,9 +26,14 @@ async fn main() -> anyhow::Result<()> {
     // Live feed: every shard 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();
+
     // Website-facing HTTP server.
     let web_state = web::AppState {
         events: bcast_tx.clone(),
+        shard: handle.clone(),
+        rpc: rpc.clone(),
     };
     tokio::spawn(async move {
         if let Err(e) = web::serve(WEB_ADDR, web_state).await {
@@ -35,11 +41,17 @@ async fn main() -> anyhow::Result<()> {
         }
     });
 
-    // Event loop: log, then broadcast. Later phases also persist to SQLite here.
+    // Event loop: a line that correlates to a pending REST call is a reply — route it to the
+    // waiting caller and stop. Everything else is a live event: log it and broadcast it.
     let feed_tx = bcast_tx.clone();
+    let route_rpc = rpc.clone();
     let mut total: u64 = 0;
     tokio::spawn(async move {
         while let Some(ev) = event_rx.recv().await {
+            if route_rpc.try_route(&ev.value).await {
+                continue; // consumed as a reply
+            }
+
             total += 1;
             match ev.kind.as_str() {
                 "server.hello" | "mob.login" | "mob.logout" | "player.death" | "vendor.sale"
@@ -49,7 +61,6 @@ async fn main() -> anyhow::Result<()> {
                 _ => tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value),
             }
 
-            // Fan out to WebSocket clients. Err just means nobody is subscribed right now.
             let _ = feed_tx.send(ev.value.to_string());
         }
     });

sidecar/src/rpc.rs

@@ -0,0 +1,99 @@
+//! Request/reply correlation over the one shard socket.
+//!
+//! REST is synchronous ("give me this character"), the shard 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.
+//!
+//! Three correlation fields are recognized, matching what the plugin echoes: `reqId` (queries —
+//! char/roster/vendor), `code` (link.confirm → link.ok/error), and `id` (towncrier). A query's
+//! `reqId` is a process-unique counter; `code`/`id` are supplied by the caller and must be unique
+//! while outstanding.
+
+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::shard::ShardHandle;
+
+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 {
+    NoShard,
+    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 shard and awaits the reply correlated by `corr_val`. The command must
+    /// already contain the correlation field (e.g. `reqId`) set to `corr_val`.
+    pub async fn call(
+        &self,
+        shard: &ShardHandle,
+        command: Value,
+        corr_val: &str,
+    ) -> Result<Value, RpcError> {
+        let (tx, rx) = oneshot::channel();
+
+        self.pending
+            .lock()
+            .await
+            .insert(corr_val.to_string(), tx);
+
+        if !shard.send(command.to_string()).await {
+            self.pending.lock().await.remove(corr_val);
+            return Err(RpcError::NoShard);
+        }
+
+        match tokio::time::timeout(REPLY_TIMEOUT, rx).await {
+            Ok(Ok(value)) => Ok(value),
+            _ => {
+                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 candidate = ["reqId", "code", "id"]
+            .iter()
+            .find_map(|k| value.get(*k).and_then(|v| v.as_str()).map(str::to_string));
+
+        let corr = match candidate {
+            Some(c) => c,
+            None => return false,
+        };
+
+        let sender = self.pending.lock().await.remove(&corr);
+        match sender {
+            Some(tx) => {
+                let _ = tx.send(value.clone());
+                true
+            }
+            None => false,
+        }
+    }
+}

sidecar/src/web.rs

@@ -1,4 +1,4 @@
-//! The website-facing HTTP surface: a WebSocket live feed now, REST queries later.
+//! The website-facing HTTP surface: a WebSocket live feed and REST queries.
 //!
 //! Unlike the shard link, this side *may* be exposed beyond loopback — it is the website's entry
 //! point, and the sidecar is the gatekeeper. It defaults to loopback anyway; widen the bind address
@@ -8,26 +8,40 @@ use std::time::Duration;
 
 use axum::{
     extract::ws::{Message, WebSocket, WebSocketUpgrade},
-    extract::State,
+    extract::{Path, State},
+    http::StatusCode,
     response::IntoResponse,
-    routing::get,
+    routing::{get, post},
-    Router,
+    Json, Router,
 };
+use serde_json::{json, Value};
 use tokio::sync::broadcast;
 use tracing::{debug, info, warn};
 
-/// Shared state handed to each request handler. `events` is the live broadcast every WebSocket
+use crate::rpc::{Rpc, RpcError};
-/// client subscribes to.
+use crate::shard::ShardHandle;
+
+/// Shared state handed to each request handler.
 #[derive(Clone)]
 pub struct AppState {
     pub events: broadcast::Sender<String>,
+    pub shard: ShardHandle,
+    pub rpc: Rpc,
 }
 
-/// Binds the website-facing HTTP server and serves until the process ends.
 pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
     let app = Router::new()
         .route("/health", get(health))
         .route("/ws", get(ws_upgrade))
+        // Queries (shard reply correlated by reqId).
+        .route("/char/:account/:slot", get(char_by_slot))
+        .route("/char/serial/:serial", get(char_by_serial))
+        .route("/roster/:account", get(roster))
+        .route("/vendors/:account", get(vendors))
+        // Inbound commands (correlated by code / id).
+        .route("/link/confirm", post(link_confirm))
+        .route("/towncrier", post(towncrier_add))
+        .route("/towncrier/:id", axum::routing::delete(towncrier_remove))
         .with_state(state);
 
     let listener = tokio::net::TcpListener::bind(addr).await?;
@@ -40,17 +54,129 @@ async fn health() -> impl IntoResponse {
     "ok"
 }
 
+// ---- shared reply handling ----
+
+/// Turns an RPC result into an HTTP response. A `bridge.error` reply from the shard becomes a 4xx;
+/// a real reply is returned as-is; transport failures map to 503/504.
+fn respond(result: Result<Value, RpcError>) -> (StatusCode, Json<Value>) {
+    match result {
+        Ok(value) => {
+            let kind = value.get("kind").and_then(|k| k.as_str()).unwrap_or("");
+            if kind == "bridge.error" || kind.ends_with(".error") {
+                let reason = value
+                    .get("reason")
+                    .and_then(|r| r.as_str())
+                    .unwrap_or("request rejected");
+                // "unknown account" / "no character" are not-founds; the rest are bad requests.
+                let code = if reason.contains("unknown") || reason.contains("no ") {
+                    StatusCode::NOT_FOUND
+                } else {
+                    StatusCode::BAD_REQUEST
+                };
+                (code, Json(value))
+            } else {
+                (StatusCode::OK, Json(value))
+            }
+        }
+        Err(RpcError::NoShard) => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            Json(json!({"error": "shard not connected"})),
+        ),
+        Err(RpcError::Timeout) => (
+            StatusCode::GATEWAY_TIMEOUT,
+            Json(json!({"error": "shard did not reply in time"})),
+        ),
+    }
+}
+
+// ---- query handlers ----
+
+async fn char_by_slot(
+    State(st): State<AppState>,
+    Path((account, slot)): Path<(String, i64)>,
+) -> impl IntoResponse {
+    let req_id = st.rpc.next_req_id();
+    let cmd = json!({"kind":"char.request","reqId":req_id,"account":account,"slot":slot});
+    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
+}
+
+async fn char_by_serial(
+    State(st): State<AppState>,
+    Path(serial): Path<String>,
+) -> impl IntoResponse {
+    let req_id = st.rpc.next_req_id();
+    let cmd = json!({"kind":"char.request","reqId":req_id,"serial":serial});
+    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
+}
+
+async fn roster(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
+    let req_id = st.rpc.next_req_id();
+    let cmd = json!({"kind":"account.roster","reqId":req_id,"account":account});
+    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
+}
+
+async fn vendors(State(st): State<AppState>, Path(account): Path<String>) -> impl IntoResponse {
+    let req_id = st.rpc.next_req_id();
+    let cmd = json!({"kind":"vendor.snapshot","reqId":req_id,"account":account});
+    respond(st.rpc.call(&st.shard, cmd, &req_id).await)
+}
+
+// ---- inbound-command handlers ----
+
+/// Body: {"code":"AB12CD","websiteUserId":"9931"}. Correlated on `code`.
+async fn link_confirm(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    let code = body.get("code").and_then(|c| c.as_str()).unwrap_or_default();
+    let web_id = body
+        .get("websiteUserId")
+        .and_then(|w| w.as_str())
+        .unwrap_or_default();
+
+    if code.is_empty() || web_id.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "code and websiteUserId are required"})),
+        );
+    }
+
+    let cmd = json!({"kind":"link.confirm","code":code,"websiteUserId":web_id});
+    let code = code.to_string();
+    respond(st.rpc.call(&st.shard, cmd, &code).await)
+}
+
+/// Body: {"id":"n123","lines":["..."],"durationSec":3600}. Correlated on `id`.
+async fn towncrier_add(State(st): State<AppState>, Json(body): Json<Value>) -> impl IntoResponse {
+    let id = body.get("id").and_then(|i| i.as_str()).unwrap_or_default();
+    if id.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "id is required"})),
+        );
+    }
+
+    let mut cmd = body.clone();
+    cmd["kind"] = json!("towncrier.add");
+    let id = id.to_string();
+    respond(st.rpc.call(&st.shard, cmd, &id).await)
+}
+
+async fn towncrier_remove(
+    State(st): State<AppState>,
+    Path(id): Path<String>,
+) -> impl IntoResponse {
+    let cmd = json!({"kind":"towncrier.remove","id":id});
+    respond(st.rpc.call(&st.shard, cmd, &id).await)
+}
+
+// ---- websocket ----
+
 async fn ws_upgrade(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
     ws.on_upgrade(move |socket| ws_client(socket, state))
 }
 
-/// One connected website client. Forwards every live event to it as a JSON text frame. A slow
-/// client that falls behind the broadcast buffer is dropped rather than allowed to stall others.
 async fn ws_client(mut socket: WebSocket, state: AppState) {
     let mut rx = state.events.subscribe();
     info!("ws client connected");
 
-    // Greet with a small hello so the client knows the feed is live.
     if socket
         .send(Message::Text(r#"{"kind":"ws.hello"}"#.to_string()))
         .await
@@ -61,37 +187,27 @@ async fn ws_client(mut socket: WebSocket, state: AppState) {
 
     loop {
         tokio::select! {
-            // Live event to push out.
             recv = rx.recv() => {
                 match recv {
                     Ok(line) => {
                         if socket.send(Message::Text(line)).await.is_err() {
-                            break; // client went away
+                            break;
                         }
                     }
                     Err(broadcast::error::RecvError::Lagged(n)) => {
                         warn!(skipped = n, "ws client lagged; dropping missed events");
-                        // keep going; the client stays live, just missed some
                     }
                     Err(broadcast::error::RecvError::Closed) => break,
                 }
             }
-            // Client -> server frames. We accept and mostly ignore them (a control channel for
-            // later); respond to pings and honor close.
             msg = socket.recv() => {
                 match msg {
                     Some(Ok(Message::Close(_))) | None => break,
-                    Some(Ok(Message::Ping(p))) => {
+                    Some(Ok(Message::Ping(p))) => { let _ = socket.send(Message::Pong(p)).await; }
-                        let _ = socket.send(Message::Pong(p)).await;
-                    }
                     Some(Ok(other)) => debug!(?other, "ws client message ignored"),
-                    Some(Err(e)) => {
+                    Some(Err(e)) => { debug!(error = %e, "ws client error"); break; }
-                        debug!(error = %e, "ws client error");
-                        break;
                 }
             }
-            }
-            // Cheap liveness ping so a dead-but-not-closed socket is noticed.
             _ = tokio::time::sleep(Duration::from_secs(30)) => {
                 if socket.send(Message::Ping(Vec::new())).await.is_err() {
                     break;