diff --git a/sidecar/README.md b/sidecar/README.md index 3db8012..e9f1ef9 100644 --- a/sidecar/README.md +++ b/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`; 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 diff --git a/sidecar/src/main.rs b/sidecar/src/main.rs index 65afe73..aa13b3f 100644 --- a/sidecar/src/main.rs +++ b/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::(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()); } }); diff --git a/sidecar/src/rpc.rs b/sidecar/src/rpc.rs new file mode 100644 index 0000000..d4b09be --- /dev/null +++ b/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>>>, + counter: Arc, +} + +#[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 { + 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, + } + } +} diff --git a/sidecar/src/web.rs b/sidecar/src/web.rs index 2fcf182..06d2fcc 100644 --- a/sidecar/src/web.rs +++ b/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, - Router, + routing::{get, post}, + 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 -/// client subscribes to. +use crate::rpc::{Rpc, RpcError}; +use crate::shard::ShardHandle; + +/// Shared state handed to each request handler. #[derive(Clone)] pub struct AppState { pub events: broadcast::Sender, + 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) -> (StatusCode, Json) { + 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, + 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, + Path(serial): Path, +) -> 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, Path(account): Path) -> 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, Path(account): Path) -> 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, Json(body): Json) -> 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, Json(body): Json) -> 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, + Path(id): Path, +) -> 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) -> 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))) => { - let _ = socket.send(Message::Pong(p)).await; - } + Some(Ok(Message::Ping(p))) => { let _ = socket.send(Message::Pong(p)).await; } Some(Ok(other)) => debug!(?other, "ws client message ignored"), - Some(Err(e)) => { - debug!(error = %e, "ws client error"); - break; - } + Some(Err(e)) => { 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;