Sidecar: WebSocket live feed

web.rs adds the website-facing HTTP surface (axum): GET /health and GET /ws.
Every shard event is broadcast to all connected WebSocket clients as a JSON text
frame. main.rs's event loop now broadcast::sends each event after logging it.

A lagging client is warned and kept live (misses events) rather than stalling
the others; a dead-but-not-closed socket is caught by a 30s server ping. The feed
is live-only -- no replay -- since history belongs to REST + SQLite. This side may
be exposed beyond loopback (it is the gatekeeper); it defaults to 127.0.0.1:8080
and wants auth before going public.

Verified end to end: a WebSocket client connected to /ws, received ws.hello, then
live pong events relayed from the shard through the shard-link -> broadcast -> ws
path -- the same path a login, sale, or IDOC alert will take to a browser.

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

sidecar/src/main.rs

@@ -1,45 +1,60 @@
 //! uo-link sidecar.
 //!
-//! Terminates the loopback link to the ServUO shard and (in later phases) exposes WebSocket + REST
-//! to the website. This first cut proves the shard link: it receives real events and can send
-//! commands back.
+//! 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 shard;
+mod web;
 
-use tokio::sync::mpsc;
+use tokio::sync::{broadcast, mpsc};
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 
 const SHARD_ADDR: &str = "127.0.0.1:7788";
+const WEB_ADDR: &str = "127.0.0.1:8080";
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     init_tracing();
-
     info!("uo-link sidecar starting");
 
+    // Shard link: events in, commands out.
     let (event_tx, mut event_rx) = mpsc::unbounded_channel::<shard::ShardEvent>();
     let handle = shard::serve(SHARD_ADDR, event_tx).await?;
 
-    // Phase 1: log every event and tally by kind. Later phases fan this out to WS + SQLite.
+    // Live feed: every shard event fans out to all connected website WebSocket clients.
+    let (bcast_tx, _) = broadcast::channel::<String>(1024);
+
+    // Website-facing HTTP server.
+    let web_state = web::AppState {
+        events: bcast_tx.clone(),
+    };
+    tokio::spawn(async move {
+        if let Err(e) = web::serve(WEB_ADDR, web_state).await {
+            tracing::error!(error = %e, "web server exited");
+        }
+    });
+
+    // Event loop: log, then broadcast. Later phases also persist to SQLite here.
+    let feed_tx = bcast_tx.clone();
     let mut total: u64 = 0;
     tokio::spawn(async move {
         while let Some(ev) = event_rx.recv().await {
             total += 1;
             match ev.kind.as_str() {
-                // These are the anchors worth surfacing at info; the rest are debug.
-                "server.hello" | "mob.login" | "mob.logout" | "player.death"
-                | "vendor.sale" | "house.decay" | "link.request" => {
+                "server.hello" | "mob.login" | "mob.logout" | "player.death" | "vendor.sale"
+                | "house.decay" | "link.request" => {
                     info!(kind = %ev.kind, n = total, "{}", ev.value);
                 }
-                _ => {
-                    tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value);
-                }
+                _ => 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());
         }
     });
 
-    // A tiny demonstration that the command path works: ping the shard once it connects.
+    // Heartbeat to the shard, exercising the command path.
     let ping_handle = handle.clone();
     tokio::spawn(async move {
         loop {
@@ -52,7 +67,6 @@ async fn main() -> anyhow::Result<()> {
         }
     });
 
-    // Run until Ctrl-C.
     tokio::signal::ctrl_c().await?;
     info!("shutting down");
     Ok(())

sidecar/src/web.rs

@@ -0,0 +1,104 @@
+//! The website-facing HTTP surface: a WebSocket live feed now, REST queries later.
+//!
+//! 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
+//! deliberately (and add auth) when the website runs on another host.
+
+use std::time::Duration;
+
+use axum::{
+    extract::ws::{Message, WebSocket, WebSocketUpgrade},
+    extract::State,
+    response::IntoResponse,
+    routing::get,
+    Router,
+};
+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.
+#[derive(Clone)]
+pub struct AppState {
+    pub events: broadcast::Sender<String>,
+}
+
+/// 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))
+        .with_state(state);
+
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    info!(%addr, "web server listening");
+    axum::serve(listener, app).await?;
+    Ok(())
+}
+
+async fn health() -> impl IntoResponse {
+    "ok"
+}
+
+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
+        .is_err()
+    {
+        return;
+    }
+
+    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
+                        }
+                    }
+                    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(other)) => debug!(?other, "ws client message ignored"),
+                    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;
+                }
+            }
+        }
+    }
+
+    info!("ws client disconnected");
+}