flatland_client_lib/

content_pack.rs

//! Sim content packs — versioned `assets/` trees published beside client gfx.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use anyhow::Context;

use crate::asset_backend::{mime_for_path, AssetBackendConfig, AssetStore};
use crate::assets::{sha256_file, AssetBundleIndex, AssetFileEntry};

/// Relative directories under `assets/` included in the sim pack (worker ContentBundle).
pub const SIM_PACK_ASSET_DIRS: &[&str] = &[
    "crafting",
    "items",
    "world",
    "config",
    "npcs",
    "loot",
    "combat",
    "dialogue",
    "quests",
    "crops",
    "gfx/sprites",
    "paperdoll",
];

/// Build a file index for sim content under `repo_root/assets/…`.
pub fn build_sim_pack_index(
    repo_root: &Path,
    publish_rev: u64,
    published_at: &str,
) -> anyhow::Result<AssetBundleIndex> {
    let assets = repo_root.join("assets");
    let mut files = BTreeMap::new();
    for rel_dir in SIM_PACK_ASSET_DIRS {
        let dir = assets.join(rel_dir);
        if dir.is_dir() {
            walk_into(&dir, &assets, &mut files)?;
        }
    }
    // Publish marker so workers know the pack rev without a separate channel.
    let marker = assets.join(".content-publish.json");
    if marker.is_file() {
        let meta = std::fs::metadata(&marker)?;
        files.insert(
            ".content-publish.json".to_string(),
            AssetFileEntry {
                sha256: sha256_file(&marker)?,
                size: meta.len(),
            },
        );
    }
    if files.is_empty() {
        anyhow::bail!(
            "sim pack is empty — expected content under {}",
            assets.display()
        );
    }
    Ok(AssetBundleIndex {
        publish_rev,
        published_at: published_at.to_string(),
        files,
    })
}

fn walk_into(
    dir: &Path,
    assets_root: &Path,
    files: &mut BTreeMap<String, AssetFileEntry>,
) -> anyhow::Result<()> {
    for entry in std::fs::read_dir(dir).with_context(|| format!("read_dir {}", dir.display()))? {
        let entry = entry?;
        let path = entry.path();
        let name = entry.file_name();
        let name_str = name.to_string_lossy();
        if name_str.starts_with('.') || name_str == "node_modules" {
            continue;
        }
        if name_str.ends_with(".map-locks.json") {
            continue;
        }
        if path.is_dir() {
            walk_into(&path, assets_root, files)?;
            continue;
        }
        if !path.is_file() {
            continue;
        }
        let rel = path
            .strip_prefix(assets_root)
            .with_context(|| format!("strip {}", path.display()))?
            .to_string_lossy()
            .replace('\\', "/");
        let meta = std::fs::metadata(&path)?;
        files.insert(
            rel,
            AssetFileEntry {
                sha256: sha256_file(&path)?,
                size: meta.len(),
            },
        );
    }
    Ok(())
}

/// Resolve an on-disk path for a sim pack relative key (`world/segments/…`).
pub fn resolve_sim_pack_file(repo_root: &Path, relative: &str) -> PathBuf {
    repo_root.join("assets").join(relative)
}

/// Upload sim pack files + `latest.json` via the configured asset store.
pub async fn upload_sim_pack(
    store: &dyn AssetStore,
    cfg: &AssetBackendConfig,
    repo_root: &Path,
    index: &AssetBundleIndex,
) -> anyhow::Result<()> {
    for (rel, _) in &index.files {
        let path = resolve_sim_pack_file(repo_root, rel);
        let bytes = std::fs::read(&path)
            .with_context(|| format!("read sim pack file {}", path.display()))?;
        let key = cfg.sim_object_key(index.publish_rev, rel);
        store
            .put(&key, mime_for_path(&path), &bytes)
            .await
            .with_context(|| format!("upload sim {key}"))?;
    }
    let latest = serde_json::to_vec(index)?;
    store
        .put(&cfg.sim_index_object(), "application/json", &latest)
        .await
        .context("upload sim latest.json")?;
    Ok(())
}

/// Download sim pack `publish_rev` into `dest_root/assets/…` (repo-shaped tree).
pub async fn download_sim_pack(
    store: &dyn AssetStore,
    cfg: &AssetBackendConfig,
    index: &AssetBundleIndex,
    dest_root: &Path,
) -> anyhow::Result<PathBuf> {
    let assets_out = dest_root.join("assets");
    std::fs::create_dir_all(&assets_out)?;

    const CONCURRENCY: usize = 8;
    let pending: Vec<(String, AssetFileEntry)> = index
        .files
        .iter()
        .map(|(k, v)| (k.clone(), v.clone()))
        .collect();
    let total = pending.len();
    let mut join_set = tokio::task::JoinSet::new();
    let mut next = 0usize;
    let done = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));

    // Local backend: read files from disk concurrently via blocking pool.
    // Remote: each task uses its own HTTP client + config clone.
    while next < pending.len() || !join_set.is_empty() {
        while join_set.len() < CONCURRENCY && next < pending.len() {
            let (rel, entry) = pending[next].clone();
            next += 1;
            let cfg = cfg.clone();
            let assets_out = assets_out.clone();
            let publish_rev = index.publish_rev;
            let done = std::sync::Arc::clone(&done);
            join_set.spawn(async move {
                let store = crate::asset_backend::store_from_config(&cfg)?;
                let key = cfg.sim_object_key(publish_rev, &rel);
                let bytes = store
                    .get(&key)
                    .await
                    .with_context(|| format!("download sim {key}"))?;
                let got = {
                    use sha2::{Digest, Sha256};
                    hex::encode(Sha256::digest(&bytes))
                };
                if got != entry.sha256 {
                    anyhow::bail!("sim pack hash mismatch for {rel}");
                }
                let dest = assets_out.join(&rel);
                if let Some(parent) = dest.parent() {
                    std::fs::create_dir_all(parent)?;
                }
                std::fs::write(&dest, &bytes)?;
                let n = done.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
                if n % 100 == 0 || n == total {
                    tracing::info!(n, total, "sim pack download progress");
                }
                Ok::<(), anyhow::Error>(())
            });
        }
        if let Some(res) = join_set.join_next().await {
            res.map_err(|e| anyhow::anyhow!("sim pack download join: {e}"))??;
        }
    }
    Ok(dest_root.to_path_buf())
}

/// Cache directory for a downloaded sim pack rev.
pub fn sim_pack_cache_dir(publish_rev: u64) -> anyhow::Result<PathBuf> {
    let base = dirs::home_dir().ok_or_else(|| anyhow::anyhow!("home directory not found"))?;
    Ok(base
        .join(".flatland3")
        .join("sim-packs")
        .join(format!("rev-{publish_rev}")))
}

/// Fetch sim `latest.json`, download pack if needed, return repo-shaped root.
///
/// Auto-heal for Oracle clone/rsync: when the checkout's `assets/.content-publish.json`
/// is ahead of (or equal to, with `FLATLAND_SIM_PACK_PREFER_REPO_ASSETS=1`) the remote
/// `latest.json`, seed `~/.flatland3/sim-packs/rev-<N>/` from local `assets/` instead of
/// re-downloading a stale pack that is missing newly cloned item templates.
pub async fn ensure_sim_pack_cached(target_rev: Option<u64>) -> anyhow::Result<(u64, PathBuf)> {
    let cfg = AssetBackendConfig::from_env();
    let store = crate::asset_backend::store_from_config(&cfg)?;
    let index_bytes = store
        .get(&cfg.sim_index_object())
        .await
        .with_context(|| format!("fetch sim latest.json ({})", cfg.sim_index_object()))?;
    let index: AssetBundleIndex = serde_json::from_slice(&index_bytes)?;
    let remote_rev = index.publish_rev;

    if let Some((local_rev, dest)) = maybe_seed_sim_pack_from_repo_assets(remote_rev)? {
        return Ok((local_rev, dest));
    }

    let rev = target_rev.unwrap_or(remote_rev);
    if rev != remote_rev {
        anyhow::bail!(
            "requested sim pack rev {rev} but remote latest is {remote_rev}"
        );
    }
    let dest = sim_pack_cache_dir(rev)?;
    if sim_pack_cache_complete(&dest, &index)? {
        return Ok((rev, dest));
    }
    // Incomplete / corrupt cache — wipe and re-download.
    let _ = std::fs::remove_dir_all(&dest);
    download_sim_pack(store.as_ref(), &cfg, &index, &dest).await?;
    if !sim_pack_cache_complete(&dest, &index)? {
        anyhow::bail!("sim pack rev {rev} incomplete after download");
    }
    Ok((rev, dest))
}

fn prefer_repo_assets_over_remote() -> bool {
    std::env::var("FLATLAND_SIM_PACK_PREFER_REPO_ASSETS")
        .map(|v| matches!(v.as_str(), "1" | "true" | "TRUE" | "yes" | "YES"))
        .unwrap_or(true)
}

/// When the repo checkout has a publish marker at/above remote latest, materialize a
/// sim-pack cache from `assets/` so cloned content (e.g. new harvest templates) wins
/// over a stale Firebase/GCS pack.
fn maybe_seed_sim_pack_from_repo_assets(
    remote_rev: u64,
) -> anyhow::Result<Option<(u64, PathBuf)>> {
    if !prefer_repo_assets_over_remote() {
        return Ok(None);
    }
    let Some(repo) = find_repo_root_for_sim_pack() else {
        return Ok(None);
    };
    let marker_path = repo.join("assets/.content-publish.json");
    let Ok(text) = std::fs::read_to_string(&marker_path) else {
        return Ok(None);
    };
    let marker: serde_json::Value = serde_json::from_str(&text)?;
    let local_rev = marker
        .get("rev")
        .and_then(|v| v.as_u64())
        .ok_or_else(|| anyhow::anyhow!("assets/.content-publish.json missing rev"))?;
    // Prefer local when ahead of remote. Equal rev still seeds so rsynced YAML that was
    // never re-uploaded (marker not bumped) replaces a stale cache built from remote.
    if local_rev < remote_rev {
        return Ok(None);
    }
    let assets = repo.join("assets");
    if !assets.join("items").is_dir() {
        return Ok(None);
    }
    let dest = sim_pack_cache_dir(local_rev)?;
    let dest_assets = dest.join("assets");
    std::fs::create_dir_all(&dest_assets)?;
    // Refresh from checkout assets (rsync-like copy of sim pack dirs + marker).
    for rel_dir in SIM_PACK_ASSET_DIRS {
        let src = assets.join(rel_dir);
        if !src.is_dir() {
            continue;
        }
        let dst = dest_assets.join(rel_dir);
        copy_dir_recursive(&src, &dst)?;
    }
    if marker_path.is_file() {
        std::fs::copy(&marker_path, dest_assets.join(".content-publish.json"))?;
    }
    // Stamp so operators can see the pack was seeded locally.
    let stamp = serde_json::json!({
        "seeded_from": assets.display().to_string(),
        "local_rev": local_rev,
        "remote_latest_seen": remote_rev,
        "reason": "repo assets at/above remote sim latest — prefer local (clone/rsync auto-heal)",
    });
    std::fs::write(
        dest.join(".seeded-from-repo.json"),
        serde_json::to_string_pretty(&stamp)?,
    )?;
    Ok(Some((local_rev, dest)))
}

fn find_repo_root_for_sim_pack() -> Option<PathBuf> {
    if let Ok(root) = std::env::var("FLATLAND_REPO_ROOT") {
        let p = PathBuf::from(root);
        if p.join("assets/items").is_dir() {
            return Some(p);
        }
    }
    // Worker cwd is typically the checkout (Oracle: /home/ubuntu/flatland).
    if let Ok(cwd) = std::env::current_dir() {
        let mut dir = cwd;
        for _ in 0..6 {
            if dir.join("assets/items").is_dir() && dir.join("assets/world").is_dir() {
                return Some(dir);
            }
            if !dir.pop() {
                break;
            }
        }
    }
    None
}

fn copy_dir_recursive(src: &Path, dst: &Path) -> anyhow::Result<()> {
    std::fs::create_dir_all(dst)?;
    for entry in std::fs::read_dir(src).with_context(|| format!("read_dir {}", src.display()))? {
        let entry = entry?;
        let name = entry.file_name();
        let name_str = name.to_string_lossy();
        if name_str.starts_with('.') || name_str == "node_modules" {
            continue;
        }
        if name_str.ends_with(".map-locks.json") {
            continue;
        }
        let from = entry.path();
        let to = dst.join(&name);
        if from.is_dir() {
            copy_dir_recursive(&from, &to)?;
        } else if from.is_file() {
            if let Some(parent) = to.parent() {
                std::fs::create_dir_all(parent)?;
            }
            std::fs::copy(&from, &to)
                .with_context(|| format!("copy {} → {}", from.display(), to.display()))?;
        }
    }
    Ok(())
}

fn sim_pack_cache_complete(dest_root: &Path, index: &AssetBundleIndex) -> anyhow::Result<bool> {
    let assets = dest_root.join("assets");
    if !assets.is_dir() {
        return Ok(false);
    }
    for (rel, entry) in &index.files {
        let path = assets.join(rel);
        if !path.is_file() {
            return Ok(false);
        }
        let meta = std::fs::metadata(&path)?;
        if meta.len() != entry.size && entry.size > 0 {
            // Size mismatch — treat as incomplete (hash check is expensive for every boot).
            return Ok(false);
        }
    }
    Ok(true)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::asset_backend::LocalAssetStore;

    #[tokio::test]
    async fn local_sim_pack_roundtrip() {
        let repo = match flatland_repo_fixture() {
            Some(p) => p,
            None => return,
        };
        let tmp =
            std::env::temp_dir().join(format!("flatland-sim-pack-test-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&tmp);
        std::fs::create_dir_all(&tmp).unwrap();
        let store = LocalAssetStore {
            root: tmp.join("store"),
        };
        let cfg = AssetBackendConfig {
            kind: crate::asset_backend::AssetBackendKind::Local,
            bucket: "local".into(),
            client_prefix: "client".into(),
            sim_prefix: "sim".into(),
            local_root: tmp.join("store"),
            s3_region: "us-east-1".into(),
            s3_endpoint: None,
            index_url_override: None,
        };
        let index = build_sim_pack_index(&repo, 42, "test").expect("index");
        assert!(!index.files.is_empty());
        upload_sim_pack(&store, &cfg, &repo, &index)
            .await
            .expect("upload");
        let dest = tmp.join("out");
        download_sim_pack(&store, &cfg, &index, &dest)
            .await
            .expect("download");
        assert!(dest.join("assets/world").is_dir() || dest.join("assets/items").is_dir());
        let _ = std::fs::remove_dir_all(&tmp);
    }

    fn flatland_repo_fixture() -> Option<PathBuf> {
        let mut dir = std::env::current_dir().ok()?;
        for _ in 0..8 {
            if dir.join("assets/world/segments").is_dir() {
                return Some(dir);
            }
            if !dir.pop() {
                break;
            }
        }
        None
    }
}