Sidecar: shard link (Rust)

First cut of the Rust sidecar, in sidecar/. It is the TCP listener the shard
dials out to; that asymmetry is what keeps the game unreachable from the website.

shard.rs: serve() binds 127.0.0.1:7788 and accepts shard connections in a loop,
re-accepting on disconnect. Each connection splits read/write: the reader parses
newline-JSON into ShardEvent { kind, value } and forwards over an mpsc; the writer
drains a command mpsc. ShardHandle::send posts to whichever shard is connected and
drops with a warning if none is -- a website query during an outage should fail
fast, not queue; live events that must survive an outage are buffered by the shard.

main.rs wires it up, logs events by kind, and runs a 15s heartbeat ping to
exercise the command path. Later phases fan events out to a WebSocket broadcaster
and SQLite, and turn REST calls into shard commands.

Verified against the live shard: the sidecar received the shard's server.hello
(parsed, fields intact), round-tripped its heartbeat ping -> pong, and after a
sidecar restart the shard reconnected on its own and re-sent hello. Tokio + serde;
axum/sqlx/tungstenite come with the WS and REST phases.

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

sidecar/src/main.rs

@@ -0,0 +1,67 @@
+//! 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.
+
+mod shard;
+
+use tokio::sync::mpsc;
+use tracing::info;
+use tracing_subscriber::EnvFilter;
+
+const SHARD_ADDR: &str = "127.0.0.1:7788";
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    init_tracing();
+
+    info!("uo-link sidecar starting");
+
+    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.
+    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" => {
+                    info!(kind = %ev.kind, n = total, "{}", ev.value);
+                }
+                _ => {
+                    tracing::debug!(kind = %ev.kind, n = total, "{}", ev.value);
+                }
+            }
+        }
+    });
+
+    // A tiny demonstration that the command path works: ping the shard once it connects.
+    let ping_handle = handle.clone();
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(15)).await;
+            if ping_handle.is_connected().await {
+                let _ = ping_handle
+                    .send(r#"{"kind":"ping","id":"sidecar-heartbeat"}"#.to_string())
+                    .await;
+            }
+        }
+    });
+
+    // Run until Ctrl-C.
+    tokio::signal::ctrl_c().await?;
+    info!("shutting down");
+    Ok(())
+}
+
+fn init_tracing() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
+        )
+        .init();
+}

sidecar/src/shard.rs

@@ -0,0 +1,154 @@
+//! The loopback link to the ServUO shard.
+//!
+//! The shard is the TCP *client*: it dials out to us. So the sidecar owns the listener, and the
+//! shard'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. See docs/PLAN.md §2.
+//!
+//! Framing is newline-delimited JSON, bidirectional: the shard sends events, we send commands. We
+//! accept one shard connection at a time and re-accept when it drops (the shard reconnects on its
+//! own, with a bounded backoff).
+
+use std::sync::Arc;
+
+use serde_json::Value;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::TcpListener;
+use tokio::sync::{mpsc, Mutex};
+use tracing::{info, warn};
+
+/// An event line received from the shard, parsed. `kind` is lifted out for routing.
+#[derive(Debug, Clone)]
+pub struct ShardEvent {
+    pub kind: String,
+    pub value: Value,
+}
+
+/// A handle for sending command lines to the shard. Cloneable and cheap.
+///
+/// Commands are dropped (with a warning) when no shard is connected, rather than buffered: a
+/// website query that arrives during a shard outage should fail fast and be retried, not silently
+/// queue behind a reconnect. Live *events* are what must survive an outage, and those the shard
+/// buffers on its side.
+#[derive(Clone)]
+pub struct ShardHandle {
+    tx: Arc<Mutex<Option<mpsc::UnboundedSender<String>>>>,
+}
+
+impl ShardHandle {
+    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 shard 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 shard connected");
+                false
+            }
+        }
+    }
+
+    pub async fn is_connected(&self) -> bool {
+        self.tx.lock().await.is_some()
+    }
+}
+
+/// Binds the loopback listener and accepts shard 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 shard is currently connected.
+pub async fn serve(
+    addr: &str,
+    event_tx: mpsc::UnboundedSender<ShardEvent>,
+) -> std::io::Result<ShardHandle> {
+    let listener = TcpListener::bind(addr).await?;
+    info!(%addr, "shard link listening");
+
+    let handle = ShardHandle::new();
+    let accept_handle = handle.clone();
+
+    tokio::spawn(async move {
+        loop {
+            match listener.accept().await {
+                Ok((stream, peer)) => {
+                    info!(%peer, "shard connected");
+                    if let Err(e) = handle_connection(stream, &event_tx, &accept_handle).await {
+                        warn!(error = %e, "shard connection ended");
+                    } else {
+                        info!("shard disconnected");
+                    }
+                    accept_handle.set(None).await;
+                }
+                Err(e) => {
+                    warn!(error = %e, "accept failed");
+                    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+                }
+            }
+        }
+    });
+
+    Ok(handle)
+}
+
+async fn handle_connection(
+    stream: tokio::net::TcpStream,
+    event_tx: &mpsc::UnboundedSender<ShardEvent>,
+    handle: &ShardHandle,
+) -> 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 line = String::new();
+
+    loop {
+        tokio::select! {
+            // Inbound: a line from the shard.
+            result = reader.read_line(&mut line) => {
+                let n = result?;
+                if n == 0 {
+                    return Ok(()); // clean EOF: shard closed
+                }
+                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(ShardEvent { kind, value });
+                        }
+                        Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
+                    }
+                }
+                line.clear();
+            }
+            // Outbound: a command to write to the shard.
+            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
+                }
+            }
+        }
+    }
+}