use std::collections::{HashMap, HashSet};
use std::ffi::{OsStr, OsString};
use std::fs::File;
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::os::fd::{AsFd, OwnedFd};
use std::os::unix::ffi::OsStrExt;
use std::path::{Component, Path, PathBuf};

use anyhow::{Context, Result, bail};
use rustix::fs::{
    AtFlags, FileType, Gid, Mode, OFlags, ResolveFlags, Timestamps, Uid, chownat, fchmod, fchown,
    futimens, linkat, mkdirat, openat2, symlinkat, unlinkat, utimensat,
};
use rustix::io::{Errno, dup};
use rustix::time::Timespec;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tar::{EntryType, Header};

pub const POLICY_NAME: &str = "lazaret-supported-image-v1";

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct IdMapRange {
    pub container_id: u32,
    pub namespace_id: u32,
    pub size: u32,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct IdMap {
    pub uid: Vec<IdMapRange>,
    pub gid: Vec<IdMapRange>,
}

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum PreparedDestinationKind {
    Directory,
    RegularFile,
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct PreparedDestination {
    pub path: PathBuf,
    pub kind: PreparedDestinationKind,
}

#[derive(Clone, Copy, Debug)]
pub struct PreparationLimits {
    pub max_layers: usize,
    pub max_entries: u64,
    pub max_hardlinks: usize,
    pub max_file_bytes: u64,
    pub max_total_bytes: u64,
    pub max_path_bytes: usize,
    pub max_pax_records: usize,
    pub max_pax_bytes: usize,
    pub max_json_bytes: u64,
    pub max_layer_blob_bytes: u64,
    pub max_uncompressed_layer_bytes: u64,
}

impl Default for PreparationLimits {
    fn default() -> Self {
        Self {
            max_layers: 256,
            max_entries: 500_000,
            max_hardlinks: 250_000,
            max_file_bytes: 8 * 1024 * 1024 * 1024,
            max_total_bytes: 128 * 1024 * 1024 * 1024,
            max_path_bytes: 4096,
            max_pax_records: 64,
            max_pax_bytes: 4 * 1024 * 1024,
            max_json_bytes: 16 * 1024 * 1024,
            max_layer_blob_bytes: 32 * 1024 * 1024 * 1024,
            max_uncompressed_layer_bytes: 128 * 1024 * 1024 * 1024,
        }
    }
}

#[derive(Deserialize)]
struct LayoutIndex {
    manifests: Vec<Descriptor>,
}

#[derive(Deserialize)]
struct Manifest {
    config: Descriptor,
    layers: Vec<Descriptor>,
}

#[derive(Deserialize)]
struct Config {
    rootfs: Rootfs,
}

#[derive(Deserialize)]
struct Rootfs {
    diff_ids: Vec<String>,
}

#[derive(Clone, Deserialize)]
struct Descriptor {
    digest: String,
    size: u64,
    #[serde(rename = "mediaType", default)]
    media_type: String,
}

pub(crate) struct VerifiedLayout {
    pub(crate) manifest_digest: [u8; 32],
    layers: Vec<Descriptor>,
    diff_ids: Vec<String>,
}

pub(crate) enum LayoutSource<'a> {
    Ambient(&'a Path),
    Retained(&'a File),
}

impl LayoutSource<'_> {
    fn open(&self, relative: &Path) -> Result<File> {
        match self {
            Self::Ambient(root) => Ok(File::open(root.join(relative))?),
            Self::Retained(root) => Ok(File::from(openat2(
                root,
                relative,
                OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                Mode::empty(),
                secure_resolve(),
            )?)),
        }
    }
}

#[derive(Default)]
struct Usage {
    entries: u64,
    hardlinks: usize,
    total_bytes: u64,
}

pub(crate) fn verify_layout(
    layout: &LayoutSource<'_>,
    limits: PreparationLimits,
) -> Result<VerifiedLayout> {
    let index: LayoutIndex = read_json(layout, Path::new("index.json"), limits.max_json_bytes)?;
    if index.manifests.len() != 1 {
        bail!("selected OCI layout must contain exactly one manifest");
    }
    let descriptor = &index.manifests[0];
    let manifest: Manifest =
        serde_json::from_slice(&verified_blob(layout, descriptor, limits.max_json_bytes)?)?;
    if manifest.layers.len() > limits.max_layers {
        bail!("policy {POLICY_NAME} rejects excessive layer count");
    }
    let config: Config = serde_json::from_slice(&verified_blob(
        layout,
        &manifest.config,
        limits.max_json_bytes,
    )?)?;
    if manifest.layers.len() != config.rootfs.diff_ids.len() {
        bail!("OCI config diff-ID count does not match layer count");
    }
    for layer in &manifest.layers {
        validate_media_type(&layer.media_type)?;
        verify_blob_digest(layout, layer, limits.max_layer_blob_bytes)?;
    }
    Ok(VerifiedLayout {
        manifest_digest: parse_sha256(&descriptor.digest)?,
        layers: manifest.layers,
        diff_ids: config.rootfs.diff_ids,
    })
}

pub(crate) fn apply_layout(
    layout: &LayoutSource<'_>,
    rootfs: &File,
    verified: &VerifiedLayout,
    id_map: &IdMap,
    limits: PreparationLimits,
) -> Result<()> {
    validate_id_map(id_map)?;
    let mut state = LayerState::default();
    let mut usage = Usage::default();
    for (index, (descriptor, diff_id)) in verified.layers.iter().zip(&verified.diff_ids).enumerate()
    {
        let spool = File::from(openat2(
            rootfs,
            ".",
            OFlags::RDWR | OFlags::TMPFILE | OFlags::CLOEXEC,
            Mode::from_raw_mode(0o600),
            secure_resolve(),
        )?);
        let observed = decompress_layer(
            layout.open(&blob_path(&descriptor.digest)?)?,
            &descriptor.media_type,
            spool.try_clone()?,
            limits.max_uncompressed_layer_bytes,
        )?;
        if &observed != diff_id {
            bail!("layer {index} diff-ID mismatch");
        }
        validate_tar(&spool, id_map, limits, &mut usage)?;
        state.apply(&spool, rootfs, id_map)?;
    }
    state.finish(rootfs)
}

pub(crate) fn materialize_destinations(
    root: &File,
    destinations: &[PreparedDestination],
    id_map: &IdMap,
) -> Result<()> {
    if destinations.is_empty() {
        return Ok(());
    }
    let uid = map_owner(0, &id_map.uid, "uid")?;
    let gid = map_owner(0, &id_map.gid, "gid")?;
    for destination in destinations {
        let relative = destination
            .path
            .strip_prefix("/")
            .context("prepared destination is not absolute")?;
        let relative = normalize_relative(relative)?;
        let (parent, leaf) = materialize_parent(root, &relative, uid, gid)?;
        let expected = match destination.kind {
            PreparedDestinationKind::Directory => FileType::Directory,
            PreparedDestinationKind::RegularFile => FileType::RegularFile,
        };
        match path_type(&parent, Path::new(&leaf))? {
            Some(actual) if actual == expected => {}
            Some(_) => bail!("prepared destination conflicts with the image layout"),
            None => match destination.kind {
                PreparedDestinationKind::Directory => {
                    mkdirat(&parent, &leaf, Mode::from_raw_mode(0o755))?;
                    set_materialized_metadata(open_component(&parent, &leaf)?, uid, gid)?;
                }
                PreparedDestinationKind::RegularFile => {
                    let file = openat2(
                        &parent,
                        &leaf,
                        OFlags::WRONLY
                            | OFlags::CREATE
                            | OFlags::EXCL
                            | OFlags::CLOEXEC
                            | OFlags::NOFOLLOW,
                        Mode::from_raw_mode(0o755),
                        secure_resolve(),
                    )?;
                    set_materialized_metadata(file, uid, gid)?;
                }
            },
        }
    }
    Ok(())
}

fn materialize_parent(root: &File, path: &Path, uid: u32, gid: u32) -> Result<(OwnedFd, OsString)> {
    let leaf = path
        .file_name()
        .context("prepared destination has no final component")?
        .to_os_string();
    let mut current = dup(root)?;
    for component in path.parent().into_iter().flat_map(Path::components) {
        let Component::Normal(name) = component else {
            bail!("prepared destination is not normalized");
        };
        match mkdirat(&current, name, Mode::from_raw_mode(0o755)) {
            Ok(()) => set_materialized_metadata(open_component(&current, name)?, uid, gid)?,
            Err(Errno::EXIST) => {}
            Err(error) => return Err(error.into()),
        }
        current = open_component(&current, name)?;
    }
    Ok((current, leaf))
}

fn set_materialized_metadata(descriptor: impl AsFd, uid: u32, gid: u32) -> Result<()> {
    fchown(
        &descriptor,
        Some(Uid::from_raw(uid)),
        Some(Gid::from_raw(gid)),
    )?;
    fchmod(&descriptor, Mode::from_raw_mode(0o755))?;
    futimens(&descriptor, &timestamps(0))?;
    Ok(())
}

fn read_json<T: serde::de::DeserializeOwned>(
    layout: &LayoutSource<'_>,
    path: &Path,
    limit: u64,
) -> Result<T> {
    let mut file = layout.open(path)?;
    let metadata = file.metadata()?;
    if metadata.len() > limit {
        bail!("OCI JSON exceeds {limit} bytes");
    }
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes)?;
    Ok(serde_json::from_slice(&bytes)?)
}

fn verified_blob(
    layout: &LayoutSource<'_>,
    descriptor: &Descriptor,
    limit: u64,
) -> Result<Vec<u8>> {
    let mut file = layout.open(&blob_path(&descriptor.digest)?)?;
    let metadata = file.metadata()?;
    if metadata.len() != descriptor.size || metadata.len() > limit {
        bail!("OCI descriptor size or resource limit mismatch");
    }
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes)?;
    if Sha256::digest(&bytes).as_slice() != parse_sha256(&descriptor.digest)? {
        bail!("OCI descriptor digest mismatch");
    }
    Ok(bytes)
}

fn verify_blob_digest(
    layout: &LayoutSource<'_>,
    descriptor: &Descriptor,
    limit: u64,
) -> Result<()> {
    let file = layout.open(&blob_path(&descriptor.digest)?)?;
    let metadata = file.metadata()?;
    if metadata.len() != descriptor.size || metadata.len() > limit {
        bail!("OCI layer size or resource limit mismatch");
    }
    let observed = digest_reader(file, limit, None)?;
    if observed != descriptor.digest {
        bail!("OCI layer digest mismatch");
    }
    Ok(())
}

fn blob_path(digest: &str) -> Result<PathBuf> {
    let component = digest
        .strip_prefix("sha256:")
        .context("only sha256 OCI descriptors are supported")?;
    if component.len() != 64 || !component.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        bail!("invalid sha256 OCI descriptor");
    }
    Ok(Path::new("blobs/sha256").join(component))
}

fn parse_sha256(digest: &str) -> Result<[u8; 32]> {
    let component = digest
        .strip_prefix("sha256:")
        .context("only sha256 OCI descriptors are supported")?;
    if component.len() != 64 {
        bail!("invalid sha256 OCI descriptor");
    }
    let mut bytes = [0; 32];
    for (index, pair) in component.as_bytes().chunks_exact(2).enumerate() {
        bytes[index] = u8::from_str_radix(std::str::from_utf8(pair)?, 16)?;
    }
    Ok(bytes)
}

fn decompress_layer(input: File, media_type: &str, output: File, limit: u64) -> Result<String> {
    let reader: Box<dyn Read> = match media_type {
        "application/vnd.oci.image.layer.v1.tar" => Box::new(input),
        "application/vnd.oci.image.layer.v1.tar+gzip" => {
            Box::new(flate2::read::GzDecoder::new(input))
        }
        "application/vnd.oci.image.layer.v1.tar+zstd" => {
            Box::new(zstd::stream::read::Decoder::new(input)?)
        }
        _ => bail!("policy {POLICY_NAME} rejects layer media type {media_type}"),
    };
    digest_reader(reader, limit, Some(output))
}

fn digest_reader(mut reader: impl Read, limit: u64, mut output: Option<File>) -> Result<String> {
    let mut hasher = Sha256::new();
    let mut buffer = vec![0; 64 * 1024];
    let mut total = 0u64;
    loop {
        let read = reader.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        total = total
            .checked_add(read as u64)
            .context("content size overflow")?;
        if total > limit {
            bail!("content exceeds {limit} bytes");
        }
        hasher.update(&buffer[..read]);
        if let Some(output) = &mut output {
            output.write_all(&buffer[..read])?;
        }
    }
    Ok(format!("sha256:{}", hex(&hasher.finalize())))
}

fn validate_tar(
    source: &File,
    id_map: &IdMap,
    limits: PreparationLimits,
    usage: &mut Usage,
) -> Result<()> {
    let mut archive = tar::Archive::new(rewound(source)?);
    let mut paths = HashSet::new();
    for entry in archive.entries()? {
        let mut entry = entry?;
        let entry_type = entry.header().entry_type();
        let path = normalize_layer_path(&entry.path()?, entry_type)?;
        if !paths.insert(path.clone()) {
            bail!("policy {POLICY_NAME} rejects duplicate layer path");
        }
        usage.entries += 1;
        if usage.entries > limits.max_entries
            || path.as_os_str().as_bytes().len() > limits.max_path_bytes
        {
            bail!("policy {POLICY_NAME} rejects entry or path resource limit");
        }
        let size = entry.header().size()?;
        usage.total_bytes = usage
            .total_bytes
            .checked_add(size)
            .context("size overflow")?;
        if size > limits.max_file_bytes || usage.total_bytes > limits.max_total_bytes {
            bail!("policy {POLICY_NAME} rejects content resource limit");
        }
        let pax: Vec<_> = entry
            .pax_extensions()?
            .into_iter()
            .flatten()
            .collect::<Result<_, _>>()?;
        let pax_bytes = pax.iter().try_fold(0usize, |total, item| {
            total
                .checked_add(item.key_bytes().len())?
                .checked_add(item.value_bytes().len())
        });
        if pax.len() > limits.max_pax_records
            || pax_bytes.is_none_or(|bytes| bytes > limits.max_pax_bytes)
        {
            bail!("policy {POLICY_NAME} rejects PAX resource limit");
        }
        for item in pax {
            if !matches!(item.key()?, "path" | "linkpath" | "mtime") {
                bail!("policy {POLICY_NAME} rejects PAX metadata");
            }
        }
        map_owner(entry.header().uid()?, &id_map.uid, "uid")?;
        map_owner(entry.header().gid()?, &id_map.gid, "gid")?;
        if !matches!(
            entry_type,
            EntryType::Regular
                | EntryType::Continuous
                | EntryType::Directory
                | EntryType::Symlink
                | EntryType::Link
        ) {
            bail!("policy {POLICY_NAME} rejects entry type {entry_type:?}");
        }
        let name = path
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or("");
        if name == ".wh." {
            bail!("whiteout has no target name");
        }
        if (name == ".wh..wh..opq" || name.starts_with(".wh."))
            && (!matches!(entry_type, EntryType::Regular | EntryType::Continuous)
                || entry.header().size()? != 0)
        {
            bail!("whiteout is not an empty regular file");
        }
        if entry_type == EntryType::Link {
            let target = entry.link_name()?.context("hardlink has no target")?;
            normalize_relative(&target)?;
            usage.hardlinks += 1;
            if usage.hardlinks > limits.max_hardlinks {
                bail!("policy {POLICY_NAME} rejects hardlink resource limit");
            }
        }
    }
    Ok(())
}

fn normalize_relative(path: &Path) -> Result<PathBuf> {
    let mut normalized = PathBuf::new();
    for component in path.components() {
        match component {
            Component::Normal(value) => normalized.push(value),
            Component::CurDir => {}
            Component::ParentDir => bail!("parent traversal is not allowed"),
            Component::RootDir | Component::Prefix(_) => bail!("absolute path is not allowed"),
        }
    }
    if normalized.as_os_str().is_empty() {
        bail!("empty path is not allowed");
    }
    Ok(normalized)
}

fn normalize_layer_path(path: &Path, entry_type: EntryType) -> Result<PathBuf> {
    match normalize_relative(path) {
        Ok(path) => Ok(path),
        Err(_)
            if entry_type == EntryType::Directory
                && path.components().next().is_some()
                && path
                    .components()
                    .all(|component| component == Component::CurDir) =>
        {
            Ok(PathBuf::from("."))
        }
        Err(error) => Err(error),
    }
}

fn map_owner(value: u64, ranges: &[IdMapRange], name: &str) -> Result<u32> {
    let Ok(value) = u32::try_from(value) else {
        bail!("{name} {value} is outside the target ID map");
    };
    for range in ranges {
        if value >= range.container_id && value - range.container_id < range.size {
            return range
                .namespace_id
                .checked_add(value - range.container_id)
                .context("mapped owner overflows");
        }
    }
    bail!("{name} {value} is outside the target ID map")
}

fn validate_id_map(id_map: &IdMap) -> Result<()> {
    for (name, ranges) in [("uid", &id_map.uid), ("gid", &id_map.gid)] {
        if ranges.is_empty() || ranges.iter().any(|range| range.size == 0) {
            bail!("{name} map is empty or invalid");
        }
        for range in ranges {
            range
                .container_id
                .checked_add(range.size - 1)
                .context("ID map overflows")?;
            range
                .namespace_id
                .checked_add(range.size - 1)
                .context("ID map overflows")?;
        }
        for (index, left) in ranges.iter().enumerate() {
            for right in &ranges[index + 1..] {
                if overlaps(left.container_id, left.size, right.container_id, right.size)
                    || overlaps(left.namespace_id, left.size, right.namespace_id, right.size)
                {
                    bail!("{name} map contains overlapping ranges");
                }
            }
        }
    }
    Ok(())
}

fn overlaps(left: u32, left_size: u32, right: u32, right_size: u32) -> bool {
    left <= right.saturating_add(right_size - 1) && right <= left.saturating_add(left_size - 1)
}

fn validate_media_type(media_type: &str) -> Result<()> {
    if matches!(
        media_type,
        "application/vnd.oci.image.layer.v1.tar"
            | "application/vnd.oci.image.layer.v1.tar+gzip"
            | "application/vnd.oci.image.layer.v1.tar+zstd"
    ) {
        Ok(())
    } else {
        bail!("policy {POLICY_NAME} rejects layer media type {media_type}")
    }
}

fn hex(bytes: &[u8]) -> String {
    use std::fmt::Write as _;

    bytes.iter().fold(
        String::with_capacity(bytes.len() * 2),
        |mut output, byte| {
            write!(output, "{byte:02x}").expect("writing to String cannot fail");
            output
        },
    )
}

#[derive(Default)]
struct LayerState {
    directories: HashMap<PathBuf, EntryMetadata>,
}

impl LayerState {
    fn apply(&mut self, tar_source: &File, root: &File, id_map: &IdMap) -> Result<()> {
        apply_whiteouts(tar_source, root, &mut self.directories)?;
        let mut archive = tar::Archive::new(rewound(tar_source)?);
        let mut hardlinks = Vec::new();
        for entry in archive.entries()? {
            let mut entry = entry?;
            let path = entry.path()?.into_owned();
            let name = path
                .file_name()
                .and_then(|value| value.to_str())
                .unwrap_or("");
            if name == ".wh..wh..opq" || name.starts_with(".wh.") {
                continue;
            }
            let metadata = EntryMetadata::from_header(path.clone(), entry.header(), id_map)?;
            match entry.header().entry_type() {
                EntryType::Regular | EntryType::Continuous => {
                    forget_directory_tree(&mut self.directories, &path);
                    remove_path(root.as_fd(), &path)?;
                    create_regular(root.as_fd(), &path, &mut entry, &metadata)?;
                }
                EntryType::Directory => {
                    ensure_directory(root.as_fd(), &path)?;
                    self.directories.insert(path, metadata);
                }
                EntryType::Symlink => {
                    let target = entry.link_name()?.context("symlink has no target")?;
                    forget_directory_tree(&mut self.directories, &path);
                    remove_path(root.as_fd(), &path)?;
                    let (parent, leaf) = ensure_parent(root.as_fd(), &path)?;
                    symlinkat(target.as_ref(), &parent, &leaf)?;
                    set_symlink_metadata(&parent, &leaf, &metadata)?;
                }
                EntryType::Link => {
                    let target = entry
                        .link_name()?
                        .context("hardlink has no target")?
                        .into_owned();
                    forget_directory_tree(&mut self.directories, &path);
                    remove_path(root.as_fd(), &path)?;
                    hardlinks.push((path, target));
                }
                entry_type => bail!("validated layer contains unsupported type {entry_type:?}"),
            }
        }
        resolve_hardlinks(root.as_fd(), hardlinks)
    }

    fn finish(mut self, root: &File) -> Result<()> {
        let mut entries: Vec<_> = self.directories.drain().map(|(_, value)| value).collect();
        entries.sort_by_key(|metadata| std::cmp::Reverse(metadata.path.components().count()));
        for metadata in entries {
            let fd = openat2(
                root.as_fd(),
                &metadata.path,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
                secure_resolve(),
            )?;
            set_fd_metadata(&fd, &metadata)?;
        }
        Ok(())
    }
}

struct EntryMetadata {
    path: PathBuf,
    uid: u32,
    gid: u32,
    mode: u32,
    mtime: u64,
}

impl EntryMetadata {
    fn from_header(path: PathBuf, header: &Header, id_map: &IdMap) -> Result<Self> {
        Ok(Self {
            path,
            uid: map_owner(header.uid()?, &id_map.uid, "uid")?,
            gid: map_owner(header.gid()?, &id_map.gid, "gid")?,
            mode: header.mode()?,
            mtime: header.mtime()?,
        })
    }
}

fn apply_whiteouts(
    tar_source: &File,
    root: &File,
    directories: &mut HashMap<PathBuf, EntryMetadata>,
) -> Result<()> {
    let mut archive = tar::Archive::new(rewound(tar_source)?);
    for entry in archive.entries()? {
        let entry = entry?;
        let path = entry.path()?.into_owned();
        let name = path
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or("");
        if name == ".wh..wh..opq" {
            if let Some(parent) = path.parent() {
                remove_children_at(root.as_fd(), parent)?;
                directories
                    .retain(|candidate, _| candidate == parent || !candidate.starts_with(parent));
            }
        } else if let Some(target_name) = name.strip_prefix(".wh.") {
            let target = path.parent().unwrap_or(Path::new("")).join(target_name);
            forget_directory_tree(directories, &target);
            remove_path(root.as_fd(), &target)?;
        }
    }
    Ok(())
}

fn forget_directory_tree(directories: &mut HashMap<PathBuf, EntryMetadata>, path: &Path) {
    directories.retain(|candidate, _| !candidate.starts_with(path));
}

fn create_regular<R: Read>(
    root: impl AsFd,
    path: &Path,
    entry: &mut tar::Entry<'_, R>,
    metadata: &EntryMetadata,
) -> Result<()> {
    let (parent, leaf) = ensure_parent(root, path)?;
    let fd = openat2(
        &parent,
        &leaf,
        OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::from_raw_mode(0o600),
        secure_resolve(),
    )?;
    let mut file = File::from(fd);
    io::copy(entry, &mut file)?;
    set_fd_metadata(&file, metadata)
}

fn ensure_directory(root: impl AsFd, path: &Path) -> Result<()> {
    if let Some(file_type) = path_type(root.as_fd(), path)? {
        if file_type == FileType::Directory {
            return Ok(());
        }
        remove_path(root.as_fd(), path)?;
    }
    let (parent, leaf) = ensure_parent(root, path)?;
    match mkdirat(&parent, &leaf, Mode::from_raw_mode(0o700)) {
        Ok(()) | Err(Errno::EXIST) => Ok(()),
        Err(error) => Err(error.into()),
    }
}

fn resolve_hardlinks(root: impl AsFd, mut pending: Vec<(PathBuf, PathBuf)>) -> Result<()> {
    while !pending.is_empty() {
        let mut remaining = Vec::new();
        let mut progress = false;
        for (path, target) in pending {
            match create_hardlink(root.as_fd(), &path, &target) {
                Ok(()) => progress = true,
                Err(error) if error.downcast_ref::<Errno>() == Some(&Errno::NOENT) => {
                    remaining.push((path, target));
                }
                Err(error) => return Err(error),
            }
        }
        if !progress {
            bail!("unresolved layer hardlinks");
        }
        pending = remaining;
    }
    Ok(())
}

fn create_hardlink(root: impl AsFd, path: &Path, target: &Path) -> Result<()> {
    let target_fd = openat2(
        root.as_fd(),
        target,
        OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
        secure_resolve(),
    )?;
    if FileType::from_raw_mode(rustix::fs::fstat(&target_fd)?.st_mode) != FileType::RegularFile {
        bail!("hardlink target is not a regular file");
    }
    let (parent, leaf) = ensure_parent(root, path)?;
    linkat(&target_fd, "", &parent, &leaf, AtFlags::EMPTY_PATH)?;
    Ok(())
}

fn path_type(root: impl AsFd, path: &Path) -> Result<Option<FileType>> {
    match openat2(
        root,
        path,
        OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
        secure_resolve(),
    ) {
        Ok(fd) => Ok(Some(FileType::from_raw_mode(
            rustix::fs::fstat(fd)?.st_mode,
        ))),
        Err(Errno::NOENT) => Ok(None),
        Err(error) => Err(error.into()),
    }
}

fn remove_path(root: impl AsFd, path: &Path) -> Result<()> {
    let Some((parent, leaf)) = existing_parent(root, path)? else {
        return Ok(());
    };
    let file_type = match openat2(
        &parent,
        &leaf,
        OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
        secure_resolve(),
    ) {
        Ok(fd) => FileType::from_raw_mode(rustix::fs::fstat(fd)?.st_mode),
        Err(Errno::NOENT) => return Ok(()),
        Err(error) => return Err(error.into()),
    };
    if file_type == FileType::Directory {
        let directory = open_component(&parent, &leaf)?;
        remove_children(&directory)?;
        unlinkat(&parent, &leaf, AtFlags::REMOVEDIR)?;
    } else {
        unlinkat(&parent, &leaf, AtFlags::empty())?;
    }
    Ok(())
}

fn remove_children_at(root: impl AsFd, path: &Path) -> Result<()> {
    if path.as_os_str().is_empty() {
        return remove_children(root);
    }
    let directory = match openat2(
        root,
        path,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
        secure_resolve(),
    ) {
        Ok(fd) => fd,
        Err(Errno::NOENT) => return Ok(()),
        Err(error) => return Err(error.into()),
    };
    remove_children(&directory)
}

pub(crate) fn remove_children(directory: impl AsFd) -> Result<()> {
    let mut buffer = [std::mem::MaybeUninit::uninit(); 8192];
    let names = {
        let mut names = Vec::new();
        let mut entries = rustix::fs::RawDir::new(directory.as_fd(), &mut buffer);
        while let Some(entry) = entries.next() {
            let entry = entry?;
            let name = entry.file_name();
            if name.to_bytes() != b"." && name.to_bytes() != b".." {
                names.push(name.to_owned());
            }
        }
        names
    };
    for name in names {
        let file_type = match openat2(
            directory.as_fd(),
            &name,
            OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
            secure_resolve(),
        ) {
            Ok(fd) => FileType::from_raw_mode(rustix::fs::fstat(fd)?.st_mode),
            Err(Errno::NOENT) => continue,
            Err(error) => return Err(error.into()),
        };
        if file_type == FileType::Directory {
            let child = open_component(directory.as_fd(), OsStr::from_bytes(name.to_bytes()))?;
            remove_children(&child)?;
            unlinkat(directory.as_fd(), name.as_c_str(), AtFlags::REMOVEDIR)?;
        } else {
            unlinkat(directory.as_fd(), name.as_c_str(), AtFlags::empty())?;
        }
    }
    Ok(())
}

fn ensure_parent(root: impl AsFd, path: &Path) -> Result<(OwnedFd, OsString)> {
    let leaf = path
        .file_name()
        .context("path has no final component")?
        .to_os_string();
    let mut current = dup(root)?;
    if let Some(parent) = path.parent() {
        for component in parent.components() {
            let Component::Normal(name) = component else {
                bail!("path is not relative");
            };
            match mkdirat(&current, name, Mode::from_raw_mode(0o700)) {
                Ok(()) | Err(Errno::EXIST) => {}
                Err(error) => return Err(error.into()),
            }
            current = open_component(&current, name)?;
        }
    }
    Ok((current, leaf))
}

fn existing_parent(root: impl AsFd, path: &Path) -> Result<Option<(OwnedFd, OsString)>> {
    let leaf = path
        .file_name()
        .context("path has no final component")?
        .to_os_string();
    let mut current = dup(root)?;
    if let Some(parent) = path.parent() {
        for component in parent.components() {
            let Component::Normal(name) = component else {
                bail!("path is not relative");
            };
            current = match open_component(&current, name) {
                Ok(fd) => fd,
                Err(Errno::NOENT) => return Ok(None),
                Err(error) => return Err(error.into()),
            };
        }
    }
    Ok(Some((current, leaf)))
}

fn open_component(parent: impl AsFd, name: &OsStr) -> rustix::io::Result<OwnedFd> {
    openat2(
        parent,
        name,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
        secure_resolve(),
    )
}

fn set_fd_metadata(fd: impl AsFd, metadata: &EntryMetadata) -> Result<()> {
    fchown(
        &fd,
        Some(Uid::from_raw(metadata.uid)),
        Some(Gid::from_raw(metadata.gid)),
    )?;
    fchmod(&fd, Mode::from_raw_mode(metadata.mode))?;
    futimens(&fd, &timestamps(metadata.mtime))?;
    Ok(())
}

fn set_symlink_metadata(parent: impl AsFd, leaf: &OsStr, metadata: &EntryMetadata) -> Result<()> {
    chownat(
        &parent,
        leaf,
        Some(Uid::from_raw(metadata.uid)),
        Some(Gid::from_raw(metadata.gid)),
        AtFlags::SYMLINK_NOFOLLOW,
    )?;
    utimensat(
        parent,
        leaf,
        &timestamps(metadata.mtime),
        AtFlags::SYMLINK_NOFOLLOW,
    )?;
    Ok(())
}

fn timestamps(seconds: u64) -> Timestamps {
    let seconds = i64::try_from(seconds).unwrap_or(i64::MAX);
    let time = Timespec {
        tv_sec: seconds,
        tv_nsec: 0,
    };
    Timestamps {
        last_access: time,
        last_modification: time,
    }
}

fn secure_resolve() -> ResolveFlags {
    ResolveFlags::BENEATH
        | ResolveFlags::NO_SYMLINKS
        | ResolveFlags::NO_MAGICLINKS
        | ResolveFlags::NO_XDEV
}

fn rewound(source: &File) -> Result<File> {
    let mut file = source.try_clone()?;
    file.seek(SeekFrom::Start(0))?;
    Ok(file)
}

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

    #[test]
    fn retained_layer_resolution_refuses_mount_crossing() {
        assert!(secure_resolve().contains(ResolveFlags::NO_XDEV));
    }
}