link/sidecar/src/main.rs

//! 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.
pub const PROTOCOL_VERSION: u32 = 2;

#[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 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");
                            }
                        }
                    }
                    _ => {}
                }
            }

            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();
}