Add central live map backend: chunk/player ingest, Vantage sidecar supervisor and proxy

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DjMbLQujBHunCCu5GpsHaT

panel/server/src/routes/livemap.rs

@@ -0,0 +1,268 @@
+//! Live map endpoints: ingest from game servers, the authenticated Vantage
+//! proxy for launchers/admin, and admin status.
+
+use super::servers::{get_server, GameServer};
+use crate::auth::{AdminUser, AuthUser};
+use crate::error::{AppError, AppResult};
+use crate::livemap::{parse_chunks, Dim, LivePlayer};
+use crate::state::AppState;
+use axum::body::{Body, Bytes};
+use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
+use axum::extract::{Path, Query, State};
+use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
+use axum::response::{IntoResponse, Response};
+use axum::Json;
+use base64::Engine;
+use serde::Deserialize;
+use serde_json::{json, Value};
+
+pub const CHUNK_INTERVAL_SECS: u64 = 20;
+pub const PLAYER_INTERVAL_MS: u64 = 1000;
+const MAX_PLAYERS: usize = 500;
+
+/// Settings handed to the game server in `/hello` and `/livemap/config`.
+pub fn game_config(enabled: bool) -> Value {
+    json!({
+        "enabled": enabled,
+        "chunk_interval_secs": CHUNK_INTERVAL_SECS,
+        "player_interval_ms": PLAYER_INTERVAL_MS,
+        "max_batch_bytes": 8 * 1024 * 1024,
+    })
+}
+
+fn require_enabled(enabled: bool) -> AppResult<()> {
+    if enabled {
+        Ok(())
+    } else {
+        Err(AppError::forbidden("Live Map is turned off for this server. Enable it under Servers in the admin panel."))
+    }
+}
+
+fn dim_of(s: &str) -> AppResult<Dim> {
+    Dim::parse(s).ok_or_else(|| AppError::bad_request("unknown dimension (use overworld, the_nether or the_end)"))
+}
+
+#[derive(Deserialize)]
+pub struct DimQuery {
+    #[serde(default = "overworld")]
+    dim: String,
+}
+fn overworld() -> String {
+    "overworld".into()
+}
+
+// ---------------------------------------------------------------------------
+// Game server → panel
+// ---------------------------------------------------------------------------
+
+pub async fn game_config_route(GameServer(server): GameServer) -> Json<Value> {
+    Json(game_config(server.live_map_enabled))
+}
+
+/// Timestamp tables of the regions the panel already has, base64 per region.
+pub async fn game_manifest(GameServer(server): GameServer, State(state): State<AppState>, Query(q): Query<DimQuery>) -> AppResult<Json<Value>> {
+    require_enabled(server.live_map_enabled)?;
+    let dim = dim_of(&q.dim)?;
+    let b64 = base64::engine::general_purpose::STANDARD;
+    let regions: Vec<Value> = state.livemap.manifest(server.id, dim).into_iter().map(|(x, z, ts)| json!({ "x": x, "z": z, "timestamps": b64.encode(ts) })).collect();
+    Ok(Json(json!({ "dimension": dim.slug(), "regions": regions })))
+}
+
+pub async fn game_chunks(GameServer(server): GameServer, State(state): State<AppState>, Query(q): Query<DimQuery>, body: Bytes) -> AppResult<Json<Value>> {
+    require_enabled(server.live_map_enabled)?;
+    let dim = dim_of(&q.dim)?;
+    let chunks = parse_chunks(&body).map_err(AppError::bad_request)?;
+    let stored = state.livemap.ingest_chunks(server.id, dim, chunks).await?;
+    Ok(Json(json!({ "stored": stored })))
+}
+
+pub async fn game_level(GameServer(server): GameServer, State(state): State<AppState>, body: Bytes) -> AppResult<Json<Value>> {
+    require_enabled(server.live_map_enabled)?;
+    // level.dat is a gzip-compressed NBT file.
+    if body.len() < 18 || body.len() > 2 * 1024 * 1024 || body[0] != 0x1f || body[1] != 0x8b {
+        return Err(AppError::bad_request("level.dat must be the gzip file from the world folder (up to 2 MiB)"));
+    }
+    state.livemap.ingest_level(server.id, body.to_vec()).await?;
+    Ok(Json(json!({ "ok": true })))
+}
+
+#[derive(Deserialize)]
+pub struct PlayersBody {
+    players: Vec<LivePlayer>,
+}
+
+fn sanitize(mut players: Vec<LivePlayer>) -> Vec<LivePlayer> {
+    players.truncate(MAX_PLAYERS);
+    players.retain(|p| p.x.is_finite() && p.y.is_finite() && p.z.is_finite() && p.x.abs() < 3e7 && p.z.abs() < 3e7);
+    for p in &mut players {
+        p.name = p.name.chars().filter(|c| !c.is_control()).take(32).collect();
+        p.uuid = p.uuid.chars().filter(|c| c.is_ascii_hexdigit() || *c == '-').take(36).collect();
+        p.dimension = p.dimension.chars().take(40).collect();
+    }
+    players
+}
+
+pub async fn game_players(GameServer(server): GameServer, State(state): State<AppState>, Json(b): Json<PlayersBody>) -> AppResult<Json<Value>> {
+    require_enabled(server.live_map_enabled)?;
+    state.livemap.set_players(server.id, sanitize(b.players));
+    Ok(Json(json!({ "ok": true })))
+}
+
+/// Streaming channel for the integration: JSON text frames carry player
+/// positions (`{"type":"players","players":[…]}`), binary frames carry chunk
+/// batches (`[dimension byte][chunk records…]`, see `parse_chunks`).
+pub async fn game_ws(GameServer(server): GameServer, State(state): State<AppState>, ws: WebSocketUpgrade) -> AppResult<Response> {
+    require_enabled(server.live_map_enabled)?;
+    Ok(ws.max_message_size(9 * 1024 * 1024).on_upgrade(move |socket| ws_loop(socket, state, server.id)))
+}
+
+async fn ws_loop(mut socket: WebSocket, state: AppState, id: i64) {
+    let mut last_check = std::time::Instant::now();
+    while let Some(Ok(msg)) = socket.recv().await {
+        if last_check.elapsed() > std::time::Duration::from_secs(30) {
+            last_check = std::time::Instant::now();
+            let enabled = get_server(&state, id).await.map(|s| s.live_map_enabled).unwrap_or(false);
+            if !enabled {
+                let _ = socket.send(Message::Text(json!({ "type": "disabled" }).to_string().into())).await;
+                break;
+            }
+        }
+        match msg {
+            Message::Text(t) => {
+                #[derive(Deserialize)]
+                struct Frame {
+                    #[serde(default)]
+                    players: Vec<LivePlayer>,
+                }
+                if let Ok(f) = serde_json::from_str::<Frame>(&t) {
+                    state.livemap.set_players(id, sanitize(f.players));
+                }
+            }
+            Message::Binary(b) => {
+                let reply = match b.split_first() {
+                    Some((d, rest)) => match Dim::ALL.get(*d as usize).copied().ok_or("unknown dimension").and_then(|dim| parse_chunks(rest).map(|c| (dim, c))) {
+                        Ok((dim, chunks)) => match state.livemap.ingest_chunks(id, dim, chunks).await {
+                            Ok(n) => json!({ "type": "stored", "count": n }),
+                            Err(e) => json!({ "type": "error", "error": e.to_string() }),
+                        },
+                        Err(e) => json!({ "type": "error", "error": e }),
+                    },
+                    None => json!({ "type": "error", "error": "empty frame" }),
+                };
+                if socket.send(Message::Text(reply.to_string().into())).await.is_err() {
+                    break;
+                }
+            }
+            Message::Ping(p) => {
+                let _ = socket.send(Message::Pong(p)).await;
+            }
+            Message::Close(_) => break,
+            _ => {}
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Launcher / admin viewers
+// ---------------------------------------------------------------------------
+
+fn label(d: Dim) -> &'static str {
+    match d {
+        Dim::Overworld => "Overworld",
+        Dim::Nether => "The Nether",
+        Dim::End => "The End",
+    }
+}
+
+async fn info(state: &AppState, id: i64, enabled: bool) -> Value {
+    let summary = state.livemap.world_summary(id);
+    let (generator, assets) = state.livemap.generator_ready();
+    let dims: Vec<Value> = summary["dimensions"]
+        .as_array()
+        .into_iter()
+        .flatten()
+        .filter_map(|d| {
+            let dim = Dim::parse(d["slug"].as_str()?)?;
+            Some(json!({ "id": dim.id(), "slug": dim.slug(), "label": label(dim), "available": d["regions"].as_u64().unwrap_or(0) > 0 }))
+        })
+        .collect();
+    let has_world = summary["level_dat"].as_bool().unwrap_or(false) && dims.iter().any(|d| d["available"] == true);
+    let message = if !enabled {
+        "Live Map is off for this server."
+    } else if !generator || !assets {
+        "The map generator isn't set up on the panel yet."
+    } else if !has_world {
+        "Waiting for the game server to upload its world."
+    } else {
+        ""
+    };
+    json!({
+        "enabled": enabled,
+        "ready": enabled && generator && assets && has_world,
+        "message": message,
+        "base": format!("/api/livemap/{id}"),
+        "dimensions": dims,
+        "players": state.livemap.live_players(id).len(),
+    })
+}
+
+/// What the launcher needs to decide whether to offer the Live Map button.
+pub async fn viewer_info(_: AuthUser, State(state): State<AppState>, Path(id): Path<i64>) -> AppResult<Json<Value>> {
+    let server = get_server(&state, id).await?;
+    Ok(Json(info(&state, id, server.live_map_enabled).await))
+}
+
+pub async fn admin_status(_: AdminUser, State(state): State<AppState>, Path(id): Path<i64>) -> AppResult<Json<Value>> {
+    let server = get_server(&state, id).await?;
+    let mut v = info(&state, id, server.live_map_enabled).await;
+    let (generator, assets) = state.livemap.generator_ready();
+    let stats = state.livemap.stats(id);
+    v["generator"] = json!({ "installed": generator, "path": state.livemap.generator_path() });
+    v["assets"] = json!({ "configured": assets, "path": state.livemap.assets_path() });
+    v["world"] = state.livemap.world_summary(id);
+    v["stats"] = serde_json::to_value(stats)?;
+    v["running"] = json!(state.livemap.sidecars_running(id).await);
+    v["last_error"] = json!(state.livemap.last_error(id));
+    v["roster"] = serde_json::to_value(state.livemap.live_players(id))?;
+    Ok(Json(v))
+}
+
+pub async fn admin_reset(_: AdminUser, State(state): State<AppState>, Path(id): Path<i64>) -> AppResult<Json<Value>> {
+    get_server(&state, id).await?;
+    state.livemap.reset(id).await?;
+    Ok(Json(json!({ "ok": true })))
+}
+
+/// Authenticated reverse proxy to the per-server Vantage sidecar. Clients
+/// use `…/api/livemap/{server}/{dimension}/v1/worlds/default/manifest.json`
+/// and everything the manifest references.
+pub async fn proxy(_: AuthUser, State(state): State<AppState>, Path((id, dim, path)): Path<(i64, String, String)>, headers: HeaderMap) -> AppResult<Response> {
+    let server = get_server(&state, id).await?;
+    require_enabled(server.live_map_enabled)?;
+    let dim = dim_of(&dim)?;
+    if path.split('/').any(|s| s == ".." || s.is_empty() || s.contains('\\')) || !path.starts_with("v1/worlds/default/") {
+        return Err(AppError::not_found("not found"));
+    }
+    if path == "v1/worlds/default/players.json" {
+        let mut resp = Json(state.livemap.players_json(id, dim)).into_response();
+        resp.headers_mut().insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
+        return Ok(resp);
+    }
+    let (port, token) = state.livemap.ensure_sidecar(id, dim).await.map_err(|m| AppError::new(StatusCode::SERVICE_UNAVAILABLE, m))?;
+    let mut req = state.http.get(format!("http://127.0.0.1:{port}/{path}")).bearer_auth(token).timeout(std::time::Duration::from_secs(120));
+    if let Some(etag) = headers.get(header::IF_NONE_MATCH) {
+        req = req.header(header::IF_NONE_MATCH, etag);
+    }
+    let upstream = req.send().await.map_err(|e| AppError::new(StatusCode::BAD_GATEWAY, format!("map generator unreachable: {e}")))?;
+    let mut out = Response::builder().status(upstream.status());
+    for name in [header::CONTENT_TYPE, header::ETAG, header::CACHE_CONTROL, header::LAST_MODIFIED] {
+        if let Some(v) = upstream.headers().get(&name) {
+            out = out.header(name, v);
+        }
+    }
+    // Tiles are immutable per etag; make sure shared caches never keep them.
+    Ok(out
+        .header(header::VARY, "Authorization")
+        .body(Body::from_stream(upstream.bytes_stream()))
+        .map_err(|e| AppError::new(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?)
+}

panel/server/src/routes/mod.rs

@@ -6,6 +6,7 @@ pub mod connections;
 pub mod economy;
 pub mod guilds;
 pub mod landing;
+pub mod livemap;
 pub mod leveling;
 pub mod meta;
 pub mod public;
@@ -45,6 +46,7 @@ pub fn api(state: &AppState) -> Router<AppState> {
         .route("/avatar/{id}", get(account::avatar))
         .route("/servers/public", get(servers::public_servers))
         .route("/servers/{id}/leaderboard", get(servers::public_server_leaderboard))
+        .route("/servers/{id}/livemap", get(livemap::viewer_info))
         .route("/leaderboard", get(servers::global_leaderboard))
         .route("/landing", get(landing::public_landing))
         .route("/launcher/authlib-injector.json", get(account::authlib_index))
@@ -140,6 +142,7 @@ pub fn api(state: &AppState) -> Router<AppState> {
         .route("/servers", get(servers::list).post(servers::create))
         .route("/servers/{id}", get(servers::detail).put(servers::update).delete(servers::remove))
         .route("/servers/{id}/token", post(servers::regenerate_token))
+        .route("/servers/{id}/livemap", get(livemap::admin_status).delete(livemap::admin_reset))
         .route("/users/{id}/activity", get(servers::user_activity))
         .route("/meta/minecraft", get(meta::minecraft))
         .route("/meta/loaders/{loader}", get(meta::loaders))
@@ -183,9 +186,20 @@ pub fn api(state: &AppState) -> Router<AppState> {
         .route("/economy/market/buy", post(economy::server_market_buy))
         .layer(DefaultBodyLimit::max(2 * 1024 * 1024));
 
+    // Live map ingest: chunk batches are larger than the other game calls.
+    let livemap_game = Router::new()
+        .route("/livemap/config", get(livemap::game_config_route))
+        .route("/livemap/manifest", get(livemap::game_manifest))
+        .route("/livemap/chunks", post(livemap::game_chunks))
+        .route("/livemap/level", post(livemap::game_level))
+        .route("/livemap/players", post(livemap::game_players))
+        .route("/livemap/ws", get(livemap::game_ws))
+        .layer(DefaultBodyLimit::max(10 * 1024 * 1024));
+
     Router::new()
+        .route("/api/livemap/{id}/{dim}/{*path}", get(livemap::proxy))
         .nest("/api/v1", launcher)
-        .nest("/api/server/v1", game)
+        .nest("/api/server/v1", game.merge(livemap_game))
         .nest("/api/admin", admin)
         .merge(crate::yggdrasil::routes().layer(DefaultBodyLimit::max(4 * 1024 * 1024)))
         .layer(axum::middleware::from_fn_with_state(state.clone(), activity::audit))

panel/server/src/routes/servers.rs

@@ -62,6 +62,7 @@ fn clip(s: &str, max: usize) -> String {
 pub struct ServerRow {
     pub instance_id: String,
     pub map_url: String,
+    pub live_map_enabled: bool,
     pub id: i64,
     pub name: String,
     #[serde(skip)]
@@ -168,6 +169,7 @@ pub async fn hello(
         "sync_interval_secs": SYNC_INTERVAL_SECS,
         "access": server.access,
         "require_launcher": server.require_launcher,
+        "live_map": super::livemap::game_config(server.live_map_enabled),
     })))
 }
 
@@ -835,6 +837,8 @@ pub struct ServerInput {
     require_launcher: bool,
     #[serde(default)]
     map_url: String,
+    #[serde(default)]
+    live_map_enabled: bool,
 }
 
 fn default_access() -> String {
@@ -868,7 +872,7 @@ impl ServerInput {
     }
 }
 
-async fn get_server(state: &AppState, id: i64) -> AppResult<ServerRow> {
+pub async fn get_server(state: &AppState, id: i64) -> AppResult<ServerRow> {
     sqlx::query_as("SELECT * FROM game_servers WHERE id = ?")
         .bind(id)
         .fetch_optional(&state.db)
@@ -880,7 +884,7 @@ pub async fn create(_: AdminUser, State(state): State<AppState>, Json(input): Js
     input.validate()?;
     let token = new_token();
     let id: i64 = sqlx::query_scalar(
-        "INSERT INTO game_servers (name, token_hash, token_hint, access, allowed_groups, require_launcher, created_at, instance_id, map_url) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
+        "INSERT INTO game_servers (name, token_hash, token_hint, access, allowed_groups, require_launcher, created_at, instance_id, map_url, live_map_enabled) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
     )
     .bind(input.name.trim())
     .bind(hash_token(&token))
@@ -891,6 +895,7 @@ pub async fn create(_: AdminUser, State(state): State<AppState>, Json(input): Js
     .bind(now())
     .bind(&input.instance_id)
     .bind(input.map_url.trim())
+    .bind(input.live_map_enabled)
     .fetch_one(&state.db)
     .await?;
     let server = get_server(&state, id).await?;
@@ -905,13 +910,14 @@ pub async fn update(
 ) -> AppResult<Json<ServerView>> {
     input.validate()?;
     get_server(&state, id).await?;
-    sqlx::query("UPDATE game_servers SET name = ?, access = ?, allowed_groups = ?, require_launcher = ?, instance_id = ?, map_url = ? WHERE id = ?")
+    sqlx::query("UPDATE game_servers SET name = ?, access = ?, allowed_groups = ?, require_launcher = ?, instance_id = ?, map_url = ?, live_map_enabled = ? WHERE id = ?")
         .bind(input.name.trim())
         .bind(&input.access)
         .bind(serde_json::to_string(&input.allowed_groups).unwrap_or_else(|_| "[]".into()))
         .bind(input.require_launcher)
         .bind(&input.instance_id)
         .bind(input.map_url.trim())
+        .bind(input.live_map_enabled)
         .bind(id)
         .execute(&state.db)
         .await?;
@@ -1096,6 +1102,7 @@ pub async fn public_servers(State(state): State<AppState>) -> AppResult<Json<Val
             "name": s.name,
             "instance_id": s.instance_id,
             "map_url": s.map_url,
+            "live_map": s.live_map_enabled,
             "online": is_online,
             "players_online": if is_online { s.online_count } else { 0 },
             "players_max": s.max_players,