//! Agents that run in their own user namespace.
//!
//! The operator lists them in `services.hyperhive.userns.agents` and names the
//! one host uid:gid that owns their data on disk (`dataOwner`); the unit's
//! environment carries both, so only root decides who is namespaced. A listed
//! agent starts with `--private-users=pick`, a root idmapped rather than
//! chowned, and every bind mounted `owneridmap`: the owner of the bind's
//! target inside the container, the agent user, is the data owner on disk.
//!
//! Getting there moves the agent's data once. [`activate`] chowns its `state`,
//! `harness`, `work` and `claude` dirs to the data owner, never touching
//! anything else, and records that in a marker only root can write; the next
//! call sees the marker and does nothing. `owneridmap` needs the target
//! directory to be owned by the agent when nspawn mounts, and nspawn creates a
//! missing one as root in a throwaway snapshot, so the mount points are made
//! in the template root too. The agent's socket dir lives on a tmpfs and goes
//! to the data owner at every start instead ([`owner_for_agent`]).

use std::os::unix::fs::{MetadataExt as _, lchown};
use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{AGENT_STATE_ROOT, BindMount};

use crate::files::SOCKET_DIR_ROOT;

const AGENTS_ENV: &str = "HIVE_PRIV_USERNS_AGENTS";
const DATA_OWNER_ENV: &str = "HIVE_PRIV_USERNS_DATA_OWNER";

/// One `<owner> <uid> <gid>` file per agent: the data owner its data was moved
/// to, and the agent's own ids inside the container.
const MARKER_ROOT: &str = "/var/lib/hive-priv/userns";

/// NixOS gives a normal user, which the agent is, an id from 1000 up.
const MIN_AGENT_UID: u32 = 1000;

/// The dirs under `<AGENT_STATE_ROOT>/<agent>` the agent writes into. `config`
/// is not one of them: hive-core writes it and the agent only reads it.
const DATA_DIRS: [&str; 4] = ["state", "harness", "work", "claude"];

/// Where one call looks and writes, and what it was told.
pub(crate) struct Env<'a> {
    pub(crate) state_root: &'a Path,
    /// The agent's socket dir, a bind source like the data dirs.
    pub(crate) socket_dir: &'a Path,
    pub(crate) marker_root: &'a Path,
    pub(crate) rootfs: &'a Path,
    /// The data owner when the agent is listed, else `None`.
    pub(crate) owner: Option<u32>,
    pub(crate) stopped: bool,
    /// Lowest uid taken for an agent's own: below it a dir belongs to root or
    /// to a system user such as hive-core.
    pub(crate) min_uid: u32,
    pub(crate) chown: &'a dyn Fn(&Path, u32, u32) -> std::io::Result<()>,
}

/// The data owner for `agent` under the unit's environment: `None` when the
/// agent is not listed.
///
/// # Errors
/// The agent is listed and the data owner is missing or not a plain uid.
pub(crate) fn owner_for(
    agents: Option<&str>,
    owner: Option<&str>,
    agent: &str,
) -> Result<Option<u32>> {
    if !agents.is_some_and(|list| list.split_whitespace().any(|a| a == agent)) {
        return Ok(None);
    }
    let owner = owner.with_context(|| {
        format!("{agent} is listed in {AGENTS_ENV}, but {DATA_OWNER_ENV} is not set")
    })?;
    let owner = owner
        .trim()
        .parse()
        .with_context(|| format!("{DATA_OWNER_ENV}={owner:?} is not a uid"))?;
    Ok(Some(owner))
}

/// [`owner_for`] under the unit's environment.
///
/// # Errors
/// As [`owner_for`].
pub(crate) fn owner_for_agent(agent: &str) -> Result<Option<u32>> {
    owner_for(
        std::env::var(AGENTS_ENV).ok().as_deref(),
        std::env::var(DATA_OWNER_ENV).ok().as_deref(),
        agent,
    )
}

/// Make `agent`'s next start of `container` namespaced when it is listed, and
/// say whether it is. `binds` are the ones the start will mount.
///
/// A container that is running, and whose data has not been moved yet, keeps
/// its current start: the move waits for a stopped container.
///
/// # Errors
/// The environment is inconsistent, the data owner changed since the data
/// moved, the agent was dropped from the list after it moved, or a chown or
/// mkdir fails.
pub(crate) fn activate(container: &str, agent: &str, binds: &[BindMount]) -> Result<bool> {
    let owner = owner_for_agent(agent)?;
    let rootfs = crate::container_config::container_rootfs(container);
    let stopped = crate::rootfs::unit_is_stopped(container)?;
    activate_in(
        &Env {
            state_root: Path::new(AGENT_STATE_ROOT),
            socket_dir: &Path::new(SOCKET_DIR_ROOT).join(agent),
            marker_root: Path::new(MARKER_ROOT),
            rootfs: &rootfs,
            owner,
            stopped,
            min_uid: MIN_AGENT_UID,
            chown: &|path, uid, gid| lchown(path, Some(uid), Some(gid)),
        },
        agent,
        binds,
    )
}

/// [`activate`] over an explicit [`Env`].
pub(crate) fn activate_in(env: &Env<'_>, agent: &str, binds: &[BindMount]) -> Result<bool> {
    crate::shared::validate_agent_name(agent)?;
    let marker = env.marker_root.join(agent);
    let moved = read_marker(&marker)?;
    let Some(owner) = env.owner else {
        if moved.is_some() {
            bail!(
                "{agent}: its data was moved to the userns data owner and it is no longer \
                 listed; list it again, or chown its dirs back and remove {}",
                marker.display()
            );
        }
        return Ok(false);
    };
    let ids = match moved {
        Some((was, uid, gid)) if was == owner => (uid, gid),
        Some((was, ..)) => {
            bail!("{agent}: its data is owned by {was}, not the configured data owner {owner}")
        }
        None if !env.stopped => return Ok(false),
        None => {
            let ids = move_data(env, agent, owner)?;
            write_marker(&marker, owner, ids)?;
            ids
        }
    };
    if env.stopped {
        mount_points(env, agent, binds, ids)?;
    }
    Ok(true)
}

/// `<owner> <uid> <gid>` from the marker, `None` when there is none.
fn read_marker(path: &Path) -> Result<Option<(u32, u32, u32)>> {
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e).with_context(|| format!("read {}", path.display())),
    };
    let ids: Vec<u32> = text
        .split_whitespace()
        .map(str::parse)
        .collect::<Result<_, _>>()
        .with_context(|| format!("parse {}", path.display()))?;
    let [owner, uid, gid] = ids[..] else {
        bail!("{} does not hold `<owner> <uid> <gid>`", path.display());
    };
    Ok(Some((owner, uid, gid)))
}

fn write_marker(path: &Path, owner: u32, (uid, gid): (u32, u32)) -> Result<()> {
    if let Some(dir) = path.parent() {
        std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
    }
    std::fs::write(path, format!("{owner} {uid} {gid}\n"))
        .with_context(|| format!("write {}", path.display()))
}

/// Chown the agent's data dirs to `owner:owner`; returns the agent's own uid
/// and gid, read from its `state` dir before the chown.
///
/// Only the [`DATA_DIRS`] under `<state_root>/<agent>` are touched, each only
/// if it is a real directory. The walk does not follow links or cross into
/// another filesystem, and the container is stopped while it runs.
fn move_data(env: &Env<'_>, agent: &str, owner: u32) -> Result<(u32, u32)> {
    let base = env.state_root.join(agent);
    let state = base.join("state");
    let meta =
        std::fs::symlink_metadata(&state).with_context(|| format!("stat {}", state.display()))?;
    let (uid, gid) = (meta.uid(), meta.gid());
    if !meta.is_dir() || uid < env.min_uid || uid == owner {
        bail!(
            "{agent}: {} is not a directory owned by a container user (uid {uid}); start the \
             agent once without the namespace so its user exists",
            state.display()
        );
    }
    for sub in DATA_DIRS {
        let dir = base.join(sub);
        match std::fs::symlink_metadata(&dir) {
            Ok(m) if m.is_dir() => chown_tree(env, &dir, m.dev(), owner)?,
            Ok(_) => bail!("{} is not a directory", dir.display()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e).with_context(|| format!("stat {}", dir.display())),
        }
    }
    Ok((uid, gid))
}

/// Chown everything under `top`, `top` included, to `owner:owner`.
fn chown_tree(env: &Env<'_>, top: &Path, dev: u64, owner: u32) -> Result<()> {
    let mut dirs: Vec<PathBuf> = vec![top.to_path_buf()];
    while let Some(dir) = dirs.pop() {
        (env.chown)(&dir, owner, owner).with_context(|| format!("chown {}", dir.display()))?;
        for entry in std::fs::read_dir(&dir).with_context(|| format!("read {}", dir.display()))? {
            let entry = entry?;
            let meta = entry.metadata()?;
            if meta.dev() != dev {
                continue;
            }
            if meta.is_dir() {
                dirs.push(entry.path());
            } else {
                (env.chown)(&entry.path(), owner, owner)
                    .with_context(|| format!("chown {}", entry.path().display()))?;
            }
        }
    }
    Ok(())
}

/// Create, in the template root, the mount point of each bind whose source is
/// one of the agent's data dirs or its socket dir, owned by the agent's ids
/// inside the container.
fn mount_points(
    env: &Env<'_>,
    agent: &str,
    binds: &[BindMount],
    (uid, gid): (u32, u32),
) -> Result<()> {
    let sources: Vec<PathBuf> = DATA_DIRS
        .iter()
        .map(|sub| env.state_root.join(agent).join(sub))
        .chain([env.socket_dir.to_path_buf()])
        .collect();
    for bind in binds.iter().filter(|b| {
        sources
            .iter()
            .any(|s| s.as_os_str() == b.host_path.as_str())
    }) {
        let parts: Vec<&str> = bind
            .container_path
            .split('/')
            .filter(|p| !p.is_empty())
            .collect();
        let Some(dir) = crate::rootfs::own_dir(env.rootfs, &parts, uid, gid)? else {
            bail!(
                "{}: a link or file where the mount point {} goes",
                env.rootfs.display(),
                bind.container_path
            );
        };
        (env.chown)(&dir, uid, gid).with_context(|| format!("chown {}", dir.display()))?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{Env, activate_in, owner_for};
    use crate::shared::test_support::scratch;
    use hive_priv_sock::BindMount;
    use std::cell::RefCell;
    use std::os::unix::fs::MetadataExt as _;
    use std::path::{Path, PathBuf};

    /// Not the uid the tests run as, so a chown is a change.
    const OWNER: u32 = 60000;

    fn write(path: &Path) {
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(path, "x").unwrap();
    }

    /// `<root>/agents/a` with data in each dir, `config` beside them, a link
    /// out of `work`, and `outside/o` next to the tree. Returns the root.
    fn tree() -> PathBuf {
        let root = scratch();
        let a = root.join("agents/a");
        for f in ["state/f", "harness/d/g", "work/x", "claude/c", "config/cfg"] {
            write(&a.join(f));
        }
        write(&root.join("outside/o"));
        std::os::unix::fs::symlink(root.join("outside"), a.join("work/link")).unwrap();
        root
    }

    /// Run [`activate_in`] on `root` and return its answer with every path it
    /// asked to chown, relative to `root`.
    fn run(
        root: &Path,
        owner: Option<u32>,
        stopped: bool,
        agent: &str,
        binds: &[BindMount],
    ) -> (anyhow::Result<bool>, Vec<String>) {
        let seen = RefCell::new(Vec::new());
        let chown = |p: &Path, _: u32, _: u32| {
            seen.borrow_mut()
                .push(p.strip_prefix(root).unwrap().display().to_string());
            Ok(())
        };
        let (state_root, rootfs) = (root.join("agents"), root.join("rootfs"));
        std::fs::create_dir_all(&rootfs).unwrap();
        let env = Env {
            state_root: &state_root,
            socket_dir: &root.join("run/a"),
            marker_root: &root.join("marker"),
            rootfs: &rootfs,
            owner,
            stopped,
            min_uid: 0,
            chown: &chown,
        };
        let answer = activate_in(&env, agent, binds);
        (answer, seen.into_inner())
    }

    fn bind(root: &Path, sub: &str, target: &str) -> BindMount {
        BindMount {
            host_path: root.join("agents/a").join(sub).display().to_string(),
            container_path: target.to_owned(),
            read_only: false,
        }
    }

    #[test]
    fn only_listed_agents_have_an_owner() {
        let owner = |agents, owner| owner_for(agents, owner, "a");
        assert_eq!(owner(Some("b a"), Some("60000")).unwrap(), Some(60000));
        assert_eq!(owner(Some("b c"), Some("60000")).unwrap(), None);
        assert_eq!(owner(None, None).unwrap(), None);
        assert!(owner(Some("a"), None).is_err(), "listed without an owner");
        assert!(owner(Some("a"), Some("root")).is_err(), "not a uid");
    }

    /// The move chowns the four data dirs and what is in them and nothing
    /// else: not `config`, not what a link in `work` points at.
    #[test]
    fn the_move_stays_inside_the_agents_data_dirs() {
        let root = tree();
        let (answer, mut seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.unwrap());
        seen.sort();
        let want = [
            "agents/a/claude",
            "agents/a/claude/c",
            "agents/a/harness",
            "agents/a/harness/d",
            "agents/a/harness/d/g",
            "agents/a/state",
            "agents/a/state/f",
            "agents/a/work",
            "agents/a/work/link",
            "agents/a/work/x",
        ];
        assert_eq!(seen, want);
        assert!(root.join("marker/a").exists());
    }

    /// Once the marker is there a second call chowns nothing. The first call
    /// in `the_move_stays_inside_the_agents_data_dirs` is the control that
    /// the same setup does chown.
    #[test]
    fn the_move_runs_once() {
        let root = tree();
        assert!(run(&root, Some(OWNER), true, "a", &[]).0.unwrap());
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.unwrap());
        assert!(seen.is_empty(), "{seen:?}");
    }

    /// A running container's data is left alone and the start stays as it is.
    #[test]
    fn a_running_container_is_not_moved() {
        let root = tree();
        let (answer, seen) = run(&root, Some(OWNER), false, "a", &[]);
        assert!(!answer.unwrap());
        assert!(seen.is_empty(), "{seen:?}");
        assert!(!root.join("marker/a").exists());
    }

    #[test]
    fn an_unlisted_agent_is_left_alone_and_a_moved_one_is_refused() {
        let root = tree();
        let (answer, seen) = run(&root, None, true, "a", &[]);
        assert!(!answer.unwrap() && seen.is_empty());
        assert!(
            run(&root, Some(OWNER), true, "a", &[]).0.unwrap(),
            "control"
        );
        assert!(run(&root, None, true, "a", &[]).0.is_err(), "left the list");
        assert!(
            run(&root, Some(OWNER + 1), true, "a", &[]).0.is_err(),
            "data owner changed"
        );
    }

    /// A data dir that is a link is refused before anything past it is touched.
    #[test]
    fn a_linked_data_dir_is_refused() {
        let root = tree();
        let work = root.join("agents/a/work");
        std::fs::remove_dir_all(&work).unwrap();
        std::os::unix::fs::symlink(root.join("outside"), &work).unwrap();
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &[]);
        assert!(answer.is_err());
        assert!(seen.iter().all(|p| !p.starts_with("outside")), "{seen:?}");
        assert!(!root.join("marker/a").exists());
    }

    #[test]
    fn a_path_is_not_an_agent_name() {
        let root = tree();
        let (answer, seen) = run(&root, Some(OWNER), true, "../outside", &[]);
        assert!(answer.is_err() && seen.is_empty());
    }

    /// The agent's own uid is read off its `state` dir, and root's or the
    /// data owner's is refused.
    #[test]
    fn the_state_dir_has_to_be_owned_by_a_container_user() {
        let root = tree();
        let euid = std::fs::metadata(root.join("agents/a/state"))
            .unwrap()
            .uid();
        let seen = RefCell::new(Vec::new());
        let chown = |p: &Path, _: u32, _: u32| {
            seen.borrow_mut().push(p.to_path_buf());
            Ok(())
        };
        let (state_root, rootfs) = (root.join("agents"), root.join("rootfs"));
        let mut env = Env {
            state_root: &state_root,
            socket_dir: &root.join("run/a"),
            marker_root: &root.join("marker"),
            rootfs: &rootfs,
            owner: Some(OWNER),
            stopped: true,
            min_uid: euid + 1,
            chown: &chown,
        };
        assert!(activate_in(&env, "a", &[]).is_err(), "below the minimum");
        assert!(seen.borrow().is_empty());
        env.min_uid = euid;
        env.owner = Some(euid);
        assert!(activate_in(&env, "a", &[]).is_err(), "already the owner");
        env.owner = Some(OWNER);
        assert!(activate_in(&env, "a", &[]).unwrap(), "control");
    }

    /// A mount point is made for a bind of the agent's own data dir, and for
    /// no other bind; a link in the way is refused.
    #[test]
    fn mount_points_are_made_for_the_agents_data_binds_only() {
        let root = tree();
        let binds = [
            bind(&root, "state", "/agents/a/state"),
            BindMount {
                host_path: root.join("outside").display().to_string(),
                container_path: "/agents/a/stray".to_owned(),
                read_only: false,
            },
            BindMount {
                host_path: root.join("run/a").display().to_string(),
                container_path: "/run/hive-agent/a".to_owned(),
                read_only: false,
            },
        ];
        let (answer, seen) = run(&root, Some(OWNER), true, "a", &binds);
        assert!(answer.unwrap());
        assert!(root.join("rootfs/agents/a/state").is_dir());
        assert!(root.join("rootfs/run/hive-agent/a").is_dir(), "socket dir");
        assert!(!root.join("rootfs/agents/a/stray").exists());
        assert!(
            seen.contains(&"rootfs/agents/a/state".to_owned()),
            "{seen:?}"
        );

        std::fs::remove_dir_all(root.join("rootfs/agents")).unwrap();
        std::os::unix::fs::symlink(root.join("outside"), root.join("rootfs/agents")).unwrap();
        assert!(run(&root, Some(OWNER), true, "a", &binds).0.is_err());
    }
}