Phase 1: chunk store, tile rendering pipeline, and Leaflet viewer

api: WS gateway with token auth (Postgres-backed servers table, seeded via `bun run seed`), column-granularity chunk store (hand-written SQL migrations, no drizzle-kit CLI — its config loader needs esbuild, which doesn't install cleanly here), dirty-chunk Redis stream producer with per-flush dedup, and a tile-serving route. worker: consumes the dirty-chunk stream via a proper consumer group, rasterizes each chunk's columns into a single-resolution top-down PNG (static pre-Flattening block-id palette), uploads to MinIO, and upserts the tile pointer. frontend: barebones Leaflet 2D viewer (CRS.Simple, one native zoom level) wired to /api/servers and /api/tiles. Object storage: tiles live in a dedicated `mcmapper-tiles` bucket on the existing shared MinIO instance (devstack-minio on octo-winsrv) instead of a per-stack container, via a scoped access key limited to that one bucket — see README's "Object storage" section. MINIO_SECRET_KEY is real and is deliberately not committed; docker-compose layers an untracked .env over .env.example for it. Full pipeline verified end-to-end against live containers: WS auth -> Postgres upsert -> deduped Redis dirty-chunk event -> worker rasterize -> MinIO upload -> tile fetch through the api route, including from the actual mod-side WS client (see MCMapper-Mod's matching commit).
2026-08-08 15:36:56 +02:00
parent 4c7cc26281
commit 7bed571ffa
30 changed files with 3779 additions and 84 deletions
@@ -1,8 +1,13 @@
 DATABASE_URL=postgres://mcmapper:mcmapper@postgres:5432/mcmapper
 REDIS_URL=redis://redis:6379
-MINIO_ENDPOINT=minio
-MINIO_PORT=9000
+
+# See api/.env.example for what this points at and why — same shared MinIO instance/bucket,
+# same rule: MINIO_SECRET_KEY is real, set it in an untracked `.env` next to this file, never here.
+MINIO_ENDPOINT=192.168.0.3
+MINIO_PORT=28205
+MINIO_USE_SSL=false
 MINIO_ACCESS_KEY=mcmapper
-MINIO_SECRET_KEY=mcmapper-dev-only
+MINIO_SECRET_KEY=changeme-set-in-untracked-env
+
 # cpu | gpu | hybrid — see render::backend. Only `cpu` exists so far (Phase 8 adds gpu/hybrid).
 RENDER_BACKEND=cpu
@@ -8,6 +8,12 @@ tokio = { version = "1", features = ["full"] }
 redis = { version = "0.27", features = ["tokio-comp"] }
 rayon = "1.10"
 anyhow = "1"
+sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "uuid"] }
+uuid = { version = "1", features = ["serde"] }
+image = { version = "0.25", default-features = false, features = ["png"] }
+aws-sdk-s3 = "1"
+aws-config = "1"
+aws-credential-types = "1"

 [profile.release]
 lto = true
@@ -0,0 +1,68 @@
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+pub async fn connect(database_url: &str) -> anyhow::Result<PgPool> {
+    Ok(PgPoolOptions::new().max_connections(5).connect(database_url).await?)
+}
+
+#[derive(sqlx::FromRow)]
+pub struct StoredColumn {
+    pub x: i32,
+    pub z: i32,
+    pub block_id: i32,
+    pub block_meta: i32,
+}
+
+pub async fn fetch_chunk_columns(
+    pool: &PgPool,
+    server_id: Uuid,
+    dimension: i32,
+    chunk_x: i32,
+    chunk_z: i32,
+) -> anyhow::Result<Vec<StoredColumn>> {
+    let min_x = chunk_x * 16;
+    let min_z = chunk_z * 16;
+    let rows = sqlx::query_as::<_, StoredColumn>(
+        r#"SELECT x, z, block_id, block_meta FROM chunk_columns
+           WHERE server_id = $1 AND dimension = $2
+             AND x >= $3 AND x < $3 + 16
+             AND z >= $4 AND z < $4 + 16"#,
+    )
+    .bind(server_id)
+    .bind(dimension)
+    .bind(min_x)
+    .bind(min_z)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows)
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn upsert_tile_pointer(
+    pool: &PgPool,
+    server_id: Uuid,
+    dimension: i32,
+    zoom: i32,
+    tile_x: i32,
+    tile_z: i32,
+    storage_key: &str,
+    content_hash: &str,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"INSERT INTO tile_pointers (server_id, dimension, zoom, tile_x, tile_z, storage_key, content_hash, rendered_at)
+           VALUES ($1, $2, $3, $4, $5, $6, $7, now())
+           ON CONFLICT (server_id, dimension, zoom, tile_x, tile_z)
+           DO UPDATE SET storage_key = excluded.storage_key, content_hash = excluded.content_hash, rendered_at = now()"#,
+    )
+    .bind(server_id)
+    .bind(dimension)
+    .bind(zoom)
+    .bind(tile_x)
+    .bind(tile_z)
+    .bind(storage_key)
+    .bind(content_hash)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
@@ -1,36 +1,137 @@
+mod db;
+mod palette;
 mod render;
+mod storage;

+use std::collections::hash_map::DefaultHasher;
+use std::hash::{Hash, Hasher};
+
+use redis::streams::{StreamReadOptions, StreamReadReply};
 use redis::AsyncCommands;
-use render::{CpuRenderBackend, RenderBackend};
+use render::{ColumnPixel, CpuRenderBackend, RenderBackend};
+use uuid::Uuid;

 const DIRTY_CHUNK_STREAM: &str = "mcmapper:dirty-chunks";
+const CONSUMER_GROUP: &str = "mcmapper-workers";

 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
    let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://redis:6379".into());
-    let client = redis::Client::open(redis_url)?;
-    let mut conn = client.get_multiplexed_async_connection().await?;
+    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
+    let minio_scheme = if std::env::var("MINIO_USE_SSL").as_deref() == Ok("true") {
+        "https"
+    } else {
+        "http"
+    };
+    let minio_endpoint = format!(
+        "{minio_scheme}://{}:{}",
+        std::env::var("MINIO_ENDPOINT").unwrap_or_else(|_| "minio".into()),
+        std::env::var("MINIO_PORT").unwrap_or_else(|_| "9000".into()),
+    );
+    let minio_access_key = std::env::var("MINIO_ACCESS_KEY").unwrap_or_else(|_| "mcmapper".into());
+    let minio_secret_key =
+        std::env::var("MINIO_SECRET_KEY").unwrap_or_else(|_| "mcmapper-dev-only".into());
+    let consumer_name =
+        std::env::var("HOSTNAME").unwrap_or_else(|_| format!("worker-{}", std::process::id()));

+    let pg_pool = db::connect(&database_url).await?;
+    let s3_client = storage::connect(&minio_endpoint, &minio_access_key, &minio_secret_key);
+    storage::ensure_bucket(&s3_client).await?;
+
+    let redis_client = redis::Client::open(redis_url)?;
+    let mut conn = redis_client.get_multiplexed_async_connection().await?;
+
+    // Idempotent: BUSYGROUP just means another worker instance already created it.
+    let _: redis::RedisResult<()> = conn
+        .xgroup_create_mkstream(DIRTY_CHUNK_STREAM, CONSUMER_GROUP, "$")
+        .await;
+
+    // Only `cpu` exists so far — `RENDER_BACKEND=gpu`/`hybrid` fall back with a warning until
+    // Phase 8 adds the wgpu path.
+    let requested_backend = std::env::var("RENDER_BACKEND").unwrap_or_else(|_| "cpu".into());
+    if requested_backend != "cpu" {
+        eprintln!(
+            "[worker] RENDER_BACKEND={requested_backend} not implemented yet, falling back to cpu"
+        );
+    }
    let backend: Box<dyn RenderBackend> = Box::new(CpuRenderBackend::new());
    println!("[worker] render backend: {}", backend.name());
-    println!("[worker] consuming stream '{DIRTY_CHUNK_STREAM}'");
+    println!("[worker] consuming stream '{DIRTY_CHUNK_STREAM}' as '{consumer_name}'");

-    // Phase 0 skeleton: poll the dirty-chunk stream and log entries. Consumer-group setup,
-    // actual tile rasterization and chunk meshing land in Phase 1+ (see render::RenderBackend).
    loop {
-        let entries: redis::streams::StreamReadReply = conn
-            .xread_options(
-                &[DIRTY_CHUNK_STREAM],
-                &["$"],
-                &redis::streams::StreamReadOptions::default().block(5000),
-            )
+        let opts = StreamReadOptions::default()
+            .group(CONSUMER_GROUP, &consumer_name)
+            .count(10)
+            .block(5000);
+
+        let reply: StreamReadReply = conn
+            .xread_options(&[DIRTY_CHUNK_STREAM], &[">"], &opts)
            .await
            .unwrap_or_default();

-        for key in entries.keys {
-            for id in key.ids {
-                println!("[worker] dirty chunk event {}: {:?}", id.id, id.map);
+        for key in reply.keys {
+            for entry in key.ids {
+                if let Err(err) = process_entry(&pg_pool, &s3_client, backend.as_ref(), &entry).await {
+                    eprintln!("[worker] failed to process {}: {err:#}", entry.id);
+                    continue;
+                }
+                let _: redis::RedisResult<()> =
+                    conn.xack(DIRTY_CHUNK_STREAM, CONSUMER_GROUP, &[&entry.id]).await;
            }
        }
    }
 }
+
+fn field(entry: &redis::streams::StreamId, name: &str) -> anyhow::Result<String> {
+    match entry.map.get(name) {
+        Some(redis::Value::BulkString(bytes)) => Ok(String::from_utf8(bytes.clone())?),
+        Some(other) => Ok(redis::from_redis_value::<String>(other)?),
+        None => anyhow::bail!("missing field '{name}'"),
+    }
+}
+
+async fn process_entry(
+    pool: &sqlx::PgPool,
+    s3_client: &aws_sdk_s3::Client,
+    backend: &dyn RenderBackend,
+    entry: &redis::streams::StreamId,
+) -> anyhow::Result<()> {
+    let server_id: Uuid = field(entry, "serverId")?.parse()?;
+    let dimension: i32 = field(entry, "dimension")?.parse()?;
+    let chunk_x: i32 = field(entry, "chunkX")?.parse()?;
+    let chunk_z: i32 = field(entry, "chunkZ")?.parse()?;
+
+    let columns = db::fetch_chunk_columns(pool, server_id, dimension, chunk_x, chunk_z).await?;
+    let pixels: Vec<ColumnPixel> = columns
+        .iter()
+        .map(|c| ColumnPixel {
+            local_x: (c.x - chunk_x * 16) as u8,
+            local_z: (c.z - chunk_z * 16) as u8,
+            block_id: c.block_id as u16,
+            block_meta: c.block_meta as u8,
+        })
+        .collect();
+
+    let png_bytes = backend.rasterize_tile(&pixels)?;
+
+    let mut hasher = DefaultHasher::new();
+    png_bytes.hash(&mut hasher);
+    let content_hash = format!("{:x}", hasher.finish());
+
+    let storage_key = format!("{server_id}/{dimension}/0/{chunk_x}/{chunk_z}.png");
+    storage::put_tile(s3_client, &storage_key, png_bytes).await?;
+    db::upsert_tile_pointer(
+        pool,
+        server_id,
+        dimension,
+        0,
+        chunk_x,
+        chunk_z,
+        &storage_key,
+        &content_hash,
+    )
+    .await?;
+
+    println!("[worker] rendered tile {storage_key} ({} columns)", pixels.len());
+    Ok(())
+}
@@ -0,0 +1,80 @@
+/// Static numeric-block-id -> RGB color table for pre-Flattening worlds (1.7.10/1.12.2), where
+/// a block's full identity is `(id, meta)` — see the mod's ChunkAdapter comment on why
+/// `blockStateId` for those leaves is just `(id << 4) | meta` rather than a dynamic palette.
+/// Covers common overworld terrain/ore blocks; anything unmapped falls back to a visually
+/// obvious placeholder so gaps are easy to spot while iterating rather than silently blending in.
+const UNKNOWN_COLOR: [u8; 3] = [204, 102, 204];
+const AIR_COLOR: [u8; 3] = [30, 30, 40];
+
+pub fn color_for(block_id: u16, meta: u8) -> [u8; 3] {
+    match block_id {
+        0 => AIR_COLOR,
+        1 => [125, 125, 125],  // stone
+        2 => [95, 159, 53],    // grass block
+        3 => [134, 96, 67],    // dirt
+        4 => [122, 122, 122],  // cobblestone
+        5 => match meta {
+            1 => [166, 128, 78],  // spruce planks
+            2 => [196, 179, 123], // birch planks
+            3 => [170, 122, 79],  // jungle planks
+            _ => [162, 130, 78],  // oak planks
+        },
+        7 => [40, 40, 40],                          // bedrock
+        8 | 9 => [63, 118, 228],                     // water (flowing/still)
+        10 | 11 => [207, 92, 32],                    // lava (flowing/still)
+        12 => [219, 211, 160],                        // sand
+        13 => [136, 126, 126],                        // gravel
+        14 => [252, 238, 75],                          // gold ore
+        15 => [216, 175, 147],                         // iron ore
+        16 => [77, 77, 77],                            // coal ore
+        17 | 162 => [92, 68, 41],                      // logs
+        18 | 161 => [60, 100, 40],                     // leaves
+        20 => [220, 236, 240],                          // glass
+        24 => [219, 207, 163],                          // sandstone
+        35 => wool_color(meta),
+        41 => [246, 238, 92],                            // gold block
+        42 => [220, 220, 220],                            // iron block
+        45 => [151, 96, 90],                              // bricks
+        48 => [90, 108, 90],                              // mossy cobblestone
+        49 => [24, 20, 36],                                // obsidian
+        56 => [141, 209, 202],                              // diamond ore
+        73 | 74 => [132, 32, 32],                           // redstone ore
+        78 => [240, 250, 255],                              // snow layer
+        79 => [140, 180, 230],                              // ice
+        80 => [248, 248, 248],                              // snow block
+        82 => [160, 164, 177],                              // clay
+        86 => [200, 128, 32],                               // pumpkin
+        87 => [110, 54, 48],                                // netherrack
+        88 => [84, 64, 51],                                 // soul sand
+        89 => [186, 148, 92],                               // glowstone
+        110 => [92, 84, 108],                               // mycelium
+        121 => [221, 223, 165],                             // end stone
+        123 | 124 => [171, 129, 85],                        // nether wart block-ish glow (placeholder)
+        129 => [79, 195, 161],                              // emerald ore
+        133 => [46, 190, 120],                              // emerald block
+        159 => [200, 130, 100],                             // stained clay (approx, ignores meta)
+        _ => UNKNOWN_COLOR,
+    }
+}
+
+fn wool_color(meta: u8) -> [u8; 3] {
+    match meta {
+        0 => [233, 236, 236],
+        1 => [240, 118, 19],
+        2 => [189, 68, 179],
+        3 => [107, 138, 201],
+        4 => [194, 173, 24],
+        5 => [65, 174, 56],
+        6 => [208, 132, 153],
+        7 => [64, 64, 64],
+        8 => [154, 161, 161],
+        9 => [46, 110, 137],
+        10 => [126, 61, 181],
+        11 => [46, 56, 141],
+        12 => [79, 50, 31],
+        13 => [53, 70, 27],
+        14 => [150, 52, 48],
+        15 => [25, 22, 22],
+        _ => UNKNOWN_COLOR,
+    }
+}
@@ -1,8 +1,20 @@
-use super::RenderBackend;
+use image::{ImageEncoder, Rgb, RgbImage};
+use std::io::Cursor;
+
+use super::{ColumnPixel, RenderBackend};
+use crate::palette::color_for;
+
+/// Native tile resolution (one pixel per block within a chunk) before upscaling for display.
+const CHUNK_SIZE: u32 = 16;
+/// Upscale factor applied so tiles are a reasonable size for a Leaflet `tileSize: 256` layer —
+/// Phase 1 has one native zoom level (see api's tile route), so this is purely cosmetic, not a
+/// multi-resolution pyramid (that's Phase 2's job).
+const UPSCALE: u32 = 16;

 /// CPU rendering backend, parallelized with rayon. Thread count/task-batch granularity is
 /// config-tunable (`auto` via `std::thread::available_parallelism`, or a `server`/`consumer`
-/// profile) — wired up once there's an actual render workload to tune (Phase 1+).
+/// profile) — wired up once there's a multi-tile batch workload to tune across (Phase 7);
+/// a single 16x16 tile is too small for rayon to help within itself.
 pub struct CpuRenderBackend;

 impl CpuRenderBackend {
@@ -15,4 +27,32 @@ impl RenderBackend for CpuRenderBackend {
    fn name(&self) -> &'static str {
        "cpu"
    }
+
+    fn rasterize_tile(&self, columns: &[ColumnPixel]) -> anyhow::Result<Vec<u8>> {
+        let mut base = RgbImage::from_pixel(CHUNK_SIZE, CHUNK_SIZE, Rgb([30, 30, 40]));
+        for col in columns {
+            if col.local_x as u32 >= CHUNK_SIZE || col.local_z as u32 >= CHUNK_SIZE {
+                continue;
+            }
+            let color = color_for(col.block_id, col.block_meta);
+            base.put_pixel(col.local_x as u32, col.local_z as u32, Rgb(color));
+        }
+
+        let mut upscaled = RgbImage::new(CHUNK_SIZE * UPSCALE, CHUNK_SIZE * UPSCALE);
+        for y in 0..upscaled.height() {
+            for x in 0..upscaled.width() {
+                let src = base.get_pixel(x / UPSCALE, y / UPSCALE);
+                upscaled.put_pixel(x, y, *src);
+            }
+        }
+
+        let mut bytes = Vec::new();
+        image::codecs::png::PngEncoder::new(&mut Cursor::new(&mut bytes)).write_image(
+            upscaled.as_raw(),
+            upscaled.width(),
+            upscaled.height(),
+            image::ExtendedColorType::Rgb8,
+        )?;
+        Ok(bytes)
+    }
 }
@@ -2,6 +2,14 @@ mod cpu;

 pub use cpu::CpuRenderBackend;

+/// One rendered column within a chunk, in chunk-local coordinates (0..16).
+pub struct ColumnPixel {
+    pub local_x: u8,
+    pub local_z: u8,
+    pub block_id: u16,
+    pub block_meta: u8,
+}
+
 /// Swappable rendering strategy (CPU via rayon / GPU via wgpu / hybrid) selected at startup
 /// via the `RENDER_BACKEND` env var. Only the CPU path exists so far; GPU/hybrid land in
 /// Phase 8, with automatic fallback to CPU if `gpu`/`hybrid` is requested but no compatible
@@ -9,6 +17,7 @@ pub use cpu::CpuRenderBackend;
 pub trait RenderBackend: Send + Sync {
    fn name(&self) -> &'static str;

-    // Tile rasterization and chunk meshing methods land alongside their Phase 1/2 callers —
-    // no point defining signatures for data shapes that don't exist yet.
+    /// Rasterize one chunk's worth of columns (up to 256, sparse if the chunk isn't fully
+    /// synced yet) into a single-resolution top-down PNG tile, returned as encoded bytes.
+    fn rasterize_tile(&self, columns: &[ColumnPixel]) -> anyhow::Result<Vec<u8>>;
 }
@@ -0,0 +1,38 @@
+use aws_sdk_s3::config::{BehaviorVersion, Builder, Credentials, Region};
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::Client;
+
+pub const TILE_BUCKET: &str = "mcmapper-tiles";
+
+pub fn connect(endpoint: &str, access_key: &str, secret_key: &str) -> Client {
+    let creds = Credentials::new(access_key, secret_key, None, None, "mcmapper-worker");
+    let config = Builder::new()
+        .region(Region::new("us-east-1"))
+        .endpoint_url(endpoint)
+        .credentials_provider(creds)
+        .force_path_style(true)
+        .behavior_version(BehaviorVersion::latest())
+        .build();
+    Client::from_conf(config)
+}
+
+/// Bucket creation is normally the `api` service's job (see api/src/minio.ts) — this only
+/// covers the case where `worker` starts before `api` has had a chance to.
+pub async fn ensure_bucket(client: &Client) -> anyhow::Result<()> {
+    if client.head_bucket().bucket(TILE_BUCKET).send().await.is_err() {
+        client.create_bucket().bucket(TILE_BUCKET).send().await?;
+    }
+    Ok(())
+}
+
+pub async fn put_tile(client: &Client, key: &str, bytes: Vec<u8>) -> anyhow::Result<()> {
+    client
+        .put_object()
+        .bucket(TILE_BUCKET)
+        .key(key)
+        .content_type("image/png")
+        .body(ByteStream::from(bytes))
+        .send()
+        .await?;
+    Ok(())
+}