wtclaude
8c4dc0ee93
The release workflow's first gate is `cargo fmt --check`, and the 3.0
cutover merge (2301c57) landed two rustfmt violations in the
`world.ruleset` path, so the run failed before it could build or tag:
- main.rs: the `upsert_ruleset(..)` call fits on one line
- store.rs: the `upsert_ruleset` signature does not
No behavior change — formatting only.
Co-Authored-By: Claude <noreply@anthropic.com>
332 lines
15 KiB
Rust
332 lines
15 KiB
Rust
//! uo-link sidecar.
|
|
//!
|
|
//! 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 config;
|
|
mod rpc;
|
|
mod shard;
|
|
mod store;
|
|
mod web;
|
|
|
|
use std::sync::atomic::AtomicI64;
|
|
use std::sync::Arc;
|
|
use std::time::Instant;
|
|
|
|
use tokio::sync::{broadcast, mpsc};
|
|
use tracing::info;
|
|
use tracing_subscriber::EnvFilter;
|
|
|
|
/// Wire-protocol version between the website and the sidecar. Bump this whenever an event or
|
|
/// endpoint's shape changes so a mismatched client is detected immediately (409 / health) instead
|
|
/// of failing in confusing ways.
|
|
///
|
|
/// v2 (Protocol 2.0): adds the account-provisioning verbs/endpoints (`POST /accounts/create`,
|
|
/// `DELETE /link/:account`) and their events. Outbound event kinds are additive, so a v1 website
|
|
/// keeps working against the live feed; the new *endpoints* require a v2 sidecar.
|
|
///
|
|
/// v3 (Protocol 3.0): adds `world.ruleset`, `points.board` and `vendor.listing` /
|
|
/// `vendor.listing.remove`, with the `GET /ruleset`, `/points` and `/market` reads that serve them
|
|
/// from the store. Same shape as the v2 bump — the kinds are additive, the endpoints are not — and
|
|
/// there is deliberately no feature-negotiation array: v3 implies all three kinds.
|
|
pub const PROTOCOL_VERSION: u32 = 3;
|
|
|
|
#[tokio::main]
|
|
async fn main() -> anyhow::Result<()> {
|
|
init_tracing();
|
|
info!("uo-link sidecar starting");
|
|
|
|
let cfg = config::Config::load()?;
|
|
info!(
|
|
shard = %cfg.shard.bind,
|
|
web = %cfg.web.bind,
|
|
auth = cfg.auth_required(),
|
|
"configuration loaded"
|
|
);
|
|
|
|
// Shard link: events in, commands out.
|
|
let (event_tx, mut event_rx) = mpsc::unbounded_channel::<shard::ShardEvent>();
|
|
let handle = shard::serve(&cfg.shard.bind, event_tx).await?;
|
|
|
|
// 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();
|
|
|
|
// Durable store: event history, economy series, cached profiles, link map.
|
|
let store = store::Store::open(&cfg.store.path).await?;
|
|
|
|
// Health/observability state.
|
|
let started = Instant::now();
|
|
let last_event = Arc::new(AtomicI64::new(0));
|
|
|
|
// Website-facing HTTP server.
|
|
let web_state = web::AppState {
|
|
events: bcast_tx.clone(),
|
|
shard: handle.clone(),
|
|
rpc: rpc.clone(),
|
|
store: store.clone(),
|
|
token: Arc::new(cfg.web.auth_token.clone()),
|
|
started,
|
|
last_event: last_event.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");
|
|
}
|
|
});
|
|
|
|
// 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, persist it, broadcast it.
|
|
let feed_tx = bcast_tx.clone();
|
|
let route_rpc = rpc.clone();
|
|
let event_store = store.clone();
|
|
let last_event_ts = last_event.clone();
|
|
let replay_handle = handle.clone(); // re-push external news to the shard on (re)connect
|
|
let mut total: u64 = 0;
|
|
tokio::spawn(async move {
|
|
while let Some(ev) = event_rx.recv().await {
|
|
// Any line from the shard — including pong heartbeats — 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
|
|
}
|
|
|
|
total += 1;
|
|
match ev.kind.as_str() {
|
|
"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),
|
|
}
|
|
|
|
// Persist, then broadcast. `pong` and `ws.hello` are ephemeral chatter, not history.
|
|
if ev.kind != "pong" {
|
|
let t = ev
|
|
.value
|
|
.get("t")
|
|
.and_then(|v| v.as_i64())
|
|
.unwrap_or_else(now_ms);
|
|
let text = ev.value.to_string();
|
|
if let Err(e) = event_store.insert_event(t, &ev.kind, &text).await {
|
|
tracing::warn!(error = %e, "failed to persist event");
|
|
}
|
|
|
|
// The champ board is a live projection: champ.update folds in the latest state (one
|
|
// row per spawn), champ.remove drops a spawn that despawned or was slain.
|
|
match ev.kind.as_str() {
|
|
"champ.update" => {
|
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
|
if let Err(e) = event_store
|
|
.upsert_champ(
|
|
serial,
|
|
ev.value.get("status").and_then(|s| s.as_str()),
|
|
ev.value.get("name").and_then(|n| n.as_str()),
|
|
&text,
|
|
t,
|
|
)
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "failed to upsert champ board");
|
|
}
|
|
}
|
|
}
|
|
"champ.remove" => {
|
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
|
if let Err(e) = event_store.delete_champ(serial).await {
|
|
tracing::warn!(error = %e, "failed to remove champ board row");
|
|
}
|
|
}
|
|
}
|
|
// Guild board (Protocol 2.0): guild.update folds in the latest roster (one row
|
|
// per guild id); guild.remove drops a disbanded guild.
|
|
"guild.update" => {
|
|
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
|
if let Err(e) = event_store
|
|
.upsert_guild(
|
|
id,
|
|
ev.value.get("name").and_then(|n| n.as_str()),
|
|
&text,
|
|
t,
|
|
)
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "failed to upsert guild board");
|
|
}
|
|
}
|
|
}
|
|
"guild.remove" => {
|
|
if let Some(id) = ev.value.get("id").and_then(|v| v.as_i64()) {
|
|
if let Err(e) = event_store.delete_guild(id).await {
|
|
tracing::warn!(error = %e, "failed to remove guild board row");
|
|
}
|
|
}
|
|
}
|
|
// Governor board (Protocol 2.0): city.update folds in each city's latest
|
|
// governance state (one row per city).
|
|
"city.update" => {
|
|
if let Some(city) = ev.value.get("city").and_then(|c| c.as_str()) {
|
|
if let Err(e) = event_store.upsert_governor(city, &text, t).await {
|
|
tracing::warn!(error = %e, "failed to upsert governor board");
|
|
}
|
|
}
|
|
}
|
|
// House registry (Protocol 2.0): house.update folds in each house's latest state
|
|
// (one row per serial); house.remove drops a demolished/traded house.
|
|
"house.update" => {
|
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
|
if let Err(e) = event_store
|
|
.upsert_house(
|
|
serial,
|
|
ev.value.get("name").and_then(|n| n.as_str()),
|
|
&text,
|
|
t,
|
|
)
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "failed to upsert house registry");
|
|
}
|
|
}
|
|
}
|
|
"house.remove" => {
|
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
|
if let Err(e) = event_store.delete_house(serial).await {
|
|
tracing::warn!(error = %e, "failed to remove house registry row");
|
|
}
|
|
}
|
|
}
|
|
// Points/loyalty boards (Protocol 3.0): one row per point system, keyed by
|
|
// the shard's own PointsType name. The plugin only emits a system whose top N
|
|
// actually moved, so this is a sparse stream of overwrites — and there is no
|
|
// `points.remove` to handle, because the shard's set of systems is fixed at
|
|
// startup and cannot shrink.
|
|
"points.board" => {
|
|
if let Some(system) = ev.value.get("system").and_then(|s| s.as_str()) {
|
|
if let Err(e) = event_store
|
|
.upsert_points_board(
|
|
system,
|
|
ev.value.get("nameString").and_then(|n| n.as_str()),
|
|
&text,
|
|
t,
|
|
)
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "failed to upsert points board");
|
|
}
|
|
}
|
|
}
|
|
// Player-vendor market index (Protocol 3.0). Each frame is authoritative for
|
|
// one vendor — the shard's round-robin sweep only emits a shop whose contents,
|
|
// prices or location actually moved — so this is a whole-row overwrite.
|
|
//
|
|
// Unlike the boards above there IS a remove: a vendor is dismissed, expires, or
|
|
// its owner switches off the in-game Vendor Search flag, and any of those must
|
|
// take the shop off the site. The last of the three is a privacy control, so
|
|
// dropping the row promptly is the point rather than housekeeping.
|
|
"vendor.listing" => {
|
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
|
let loc = ev.value.get("location");
|
|
let field = |k: &str| loc.and_then(|l| l.get(k));
|
|
if let Err(e) = event_store
|
|
.upsert_vendor(
|
|
serial,
|
|
ev.value.get("shopName").and_then(|v| v.as_str()),
|
|
ev.value.get("ownerName").and_then(|v| v.as_str()),
|
|
field("map").and_then(|v| v.as_str()),
|
|
field("x").and_then(|v| v.as_i64()),
|
|
field("y").and_then(|v| v.as_i64()),
|
|
field("region").and_then(|v| v.as_str()),
|
|
ev.value.get("count").and_then(|v| v.as_i64()),
|
|
&text,
|
|
t,
|
|
)
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "failed to upsert vendor listing");
|
|
}
|
|
}
|
|
}
|
|
"vendor.listing.remove" => {
|
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
|
if let Err(e) = event_store.delete_vendor(serial).await {
|
|
tracing::warn!(error = %e, "failed to remove vendor listing");
|
|
}
|
|
}
|
|
}
|
|
// Shard ruleset (Protocol 3.0): a singleton projection. The shard re-emits
|
|
// world.ruleset on every connect, so this row is simply overwritten; `rev`
|
|
// lets a reader tell a re-send from an actual config change.
|
|
"world.ruleset" => {
|
|
if let Err(e) = event_store
|
|
.upsert_ruleset(ev.value.get("rev").and_then(|r| r.as_str()), &text, t)
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "failed to upsert ruleset");
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
// On a shard (re)connect, re-push the stored external news: the shard rebuilds
|
|
// TownCryerSystem.NewsEntries from scratch each boot and does not persist ours. Replay
|
|
// with announce=false so a restart does not re-proclaim every article at once. news.add
|
|
// is idempotent by id, so replaying to a still-populated shard is harmless.
|
|
if ev.kind == "server.hello" {
|
|
match event_store.news_all().await {
|
|
Ok(items) => {
|
|
for mut item in items {
|
|
if let Some(obj) = item.as_object_mut() {
|
|
obj.insert("announce".to_string(), serde_json::json!(false));
|
|
}
|
|
if !replay_handle.send(item.to_string()).await {
|
|
break; // shard went away mid-replay
|
|
}
|
|
}
|
|
}
|
|
Err(e) => tracing::warn!(error = %e, "news replay: could not read stored news"),
|
|
}
|
|
}
|
|
|
|
let _ = feed_tx.send(ev.value.to_string());
|
|
}
|
|
});
|
|
|
|
// Heartbeat to the shard, exercising the command path.
|
|
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;
|
|
}
|
|
}
|
|
});
|
|
|
|
tokio::signal::ctrl_c().await?;
|
|
info!("shutting down");
|
|
Ok(())
|
|
}
|
|
|
|
fn now_ms() -> i64 {
|
|
use std::time::{SystemTime, UNIX_EPOCH};
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.map(|d| d.as_millis() as i64)
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
fn init_tracing() {
|
|
tracing_subscriber::fmt()
|
|
.with_env_filter(
|
|
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
|
|
)
|
|
.init();
|
|
}
|