lazaret/crates/lazaret-prepared-lower/src/store.rs

use std::fs::File;
use std::io;
use std::os::fd::AsFd;
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use anyhow::{Context, Result, bail};
use lazaret_core::configuration::ConfinedPathPlan;
use lazaret_core::identity::{
    DeviceId, HostPath, IdMapDigest, InodeId, PolicyId, PrepareOperationId, PreparedArtifactId,
    SelectedManifestDigest, SourceIdentity,
};
use lazaret_core::records::{ArtifactState, PrepareOperation, PrepareStageState, PreparedArtifact};
use lazaret_linux::filesystem::{ExclusiveFileLock, try_exclusive_flock};
use lazaret_storage_catalog::{Catalog, CatalogError, CatalogRow};
use rustix::fs::{
    AtFlags, Mode, OFlags, RenameFlags, ResolveFlags, mkdirat, openat2, renameat, renameat_with,
    unlinkat,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::layer::{
    IdMap, LayoutSource, POLICY_NAME, PreparationLimits, PreparedDestination,
    PreparedDestinationKind, apply_layout, materialize_destinations, remove_children,
    verify_layout,
};
use crate::transport::TransportedImage;
use crate::{PreparedLowerError, PreparedLowerResult};

const PHYSICAL_LOCK_WAIT: Duration = Duration::from_secs(5);
const LOCK_RETRY: Duration = Duration::from_millis(2);

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct PrepareRequest {
    pub operation_id: PrepareOperationId,
    pub layout_path: PathBuf,
    pub source_identity: Vec<u8>,
    pub selected_manifest_digest: [u8; 32],
    pub id_map: IdMap,
    pub destinations: Vec<PreparedDestination>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PrepareRecovery {
    ClearedUnmaterializedIntent,
    PreservedLiveOwnedStage,
    RemovedOwnedStage,
    PublishedCompletedStage,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReclaimOutcome {
    Removed,
    ReplacementPreserved,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PreparedArtifactGcResult {
    pub prepared_artifact_id: PreparedArtifactId,
    pub outcome: ReclaimOutcome,
}

#[derive(Clone, Debug)]
pub struct PreparedLowerStore {
    root: Arc<File>,
    catalog: Catalog,
    limits: PreparationLimits,
}

impl PreparedLowerStore {
    /// Creates the fixed prepared-lower directories beneath an existing Storage Root.
    ///
    /// # Errors
    ///
    /// Returns an error if the root cannot be opened or any fixed path is not a real directory.
    pub fn initialize(root: impl Into<PathBuf>, catalog: Catalog) -> PreparedLowerResult<Self> {
        Self::initialize_inner(root, catalog).map_err(PreparedLowerError::confinement)
    }

    fn initialize_inner(root: impl Into<PathBuf>, catalog: Catalog) -> Result<Self> {
        let root = root.into();
        let root = File::open(&root).context("opening Storage Root")?;
        Self::initialize_retained(root, catalog)
    }

    /// Retains the exact validated Storage Root and creates only fixed children beneath it.
    ///
    /// # Errors
    /// Returns an error if Storage Root authority drifts or a fixed child is not a real directory.
    pub fn under_storage_root(
        storage_root: &ConfinedPathPlan,
        catalog: Catalog,
    ) -> PreparedLowerResult<Self> {
        Self::under_storage_root_inner(storage_root, catalog)
            .map_err(PreparedLowerError::confinement)
    }

    fn under_storage_root_inner(storage_root: &ConfinedPathPlan, catalog: Catalog) -> Result<Self> {
        let storage_root = File::from(
            storage_root
                .open_authority()
                .context("opening validated Storage Root")?,
        );
        Self::under_storage_root_authority_inner(&storage_root, catalog)
    }

    /// Creates the prepared-lower store from an exact inherited Storage Root authority.
    ///
    /// # Errors
    /// Refuses fixed-child conflicts, changed ownership, or catalog initialization failure.
    pub fn under_storage_root_authority(
        storage_root: &File,
        catalog: Catalog,
    ) -> PreparedLowerResult<Self> {
        Self::under_storage_root_authority_inner(storage_root, catalog)
            .map_err(PreparedLowerError::confinement)
    }

    fn under_storage_root_authority_inner(storage_root: &File, catalog: Catalog) -> Result<Self> {
        match mkdirat(storage_root, "prepared-lower", Mode::from_raw_mode(0o700)) {
            Ok(()) | Err(rustix::io::Errno::EXIST) => {}
            Err(error) => return Err(error.into()),
        }
        let root = open_directory_at(storage_root, Path::new("prepared-lower"))?;
        require_owned_directory(&root)?;
        Self::initialize_retained(root, catalog)
    }

    fn initialize_retained(root: File, catalog: Catalog) -> Result<Self> {
        let store = Self {
            root: Arc::new(root),
            catalog,
            limits: PreparationLimits::default(),
        };
        for name in ["artifacts", "stages", "quarantine"] {
            match mkdirat(&store.root, name, Mode::from_raw_mode(0o700)) {
                Ok(()) | Err(rustix::io::Errno::EXIST) => {}
                Err(error) => return Err(error.into()),
            }
            let directory = open_directory_at(&store.root, Path::new(name))?;
            require_owned_directory(&directory)?;
        }
        open_lock_at(&store.root)?;
        Ok(store)
    }

    #[must_use]
    pub fn with_limits(mut self, limits: PreparationLimits) -> Self {
        self.limits = limits;
        self
    }

    /// Opens the exact published artifact beneath the retained prepared-lower root.
    ///
    /// # Errors
    /// Refuses a non-canonical artifact record, catalog drift, or physical identity drift.
    pub fn open_artifact_authority(
        &self,
        artifact: &PreparedArtifact,
    ) -> PreparedLowerResult<File> {
        self.open_artifact_authority_inner(artifact)
            .map_err(PreparedLowerError::confinement)
    }

    fn open_artifact_authority_inner(&self, artifact: &PreparedArtifact) -> Result<File> {
        let catalog_artifact = self
            .catalog_artifact(artifact.prepared_artifact_id)?
            .context("prepared artifact is absent from the catalog")?;
        if catalog_artifact != *artifact {
            bail!("prepared artifact does not match the catalog");
        }
        self.require_artifact_identity(artifact)?;
        self.open_relative(&artifact_path(artifact.prepared_artifact_id))?
            .context("prepared artifact path is absent")
    }

    /// Materializes and atomically publishes one verified immutable prepared lower.
    ///
    /// # Errors
    ///
    /// Refuses corrupt or unsupported OCI input, forged physical ownership, catalog conflicts,
    /// resource limits, and publication failures.
    pub fn prepare(&self, request: &PrepareRequest) -> PreparedLowerResult<PreparedArtifact> {
        self.prepare_from_source(request, &LayoutSource::Ambient(&request.layout_path), false)
    }

    /// Materializes from an exact inherited OCI-layout directory authority.
    ///
    /// # Errors
    /// Refuses invalid layout content, mapped ownership, catalog state, or publication.
    pub fn prepare_from_layout_authority(
        &self,
        request: &PrepareRequest,
        layout: &File,
    ) -> PreparedLowerResult<PreparedArtifact> {
        self.prepare_from_source(request, &LayoutSource::Retained(layout), false)
    }

    /// Materializes from an exact OCI layout while executing inside the target ID-map namespace.
    ///
    /// # Errors
    /// Refuses invalid layout content, logical ownership outside the target map, catalog state,
    /// or publication.
    pub fn prepare_from_layout_authority_in_target_namespace(
        &self,
        request: &PrepareRequest,
        layout: &File,
    ) -> PreparedLowerResult<PreparedArtifact> {
        self.prepare_from_source(request, &LayoutSource::Retained(layout), true)
    }

    /// Verifies and prepares an image through its retained transport-layout authority.
    ///
    /// # Errors
    /// Refuses ambient-only transported images and all normal preparation validation failures.
    pub fn prepare_transported(
        &self,
        operation_id: PrepareOperationId,
        image: &TransportedImage,
        id_map: IdMap,
        destinations: Vec<PreparedDestination>,
    ) -> PreparedLowerResult<PreparedArtifact> {
        self.prepare_transported_inner(operation_id, image, id_map, destinations)
    }

    fn prepare_transported_inner(
        &self,
        operation_id: PrepareOperationId,
        image: &TransportedImage,
        id_map: IdMap,
        destinations: Vec<PreparedDestination>,
    ) -> PreparedLowerResult<PreparedArtifact> {
        let layout = image
            .retained_layout()
            .context("transported image has no retained layout authority")
            .map_err(PreparedLowerError::confinement)?;
        let request = PrepareRequest {
            operation_id,
            layout_path: image.layout_path.clone(),
            source_identity: image.source_identity.clone(),
            selected_manifest_digest: image.selected_manifest_digest,
            id_map,
            destinations,
        };
        self.prepare_from_source(&request, &LayoutSource::Retained(layout.root()), false)
    }

    fn prepare_from_source(
        &self,
        request: &PrepareRequest,
        layout: &LayoutSource<'_>,
        materialize_logical_owners: bool,
    ) -> PreparedLowerResult<PreparedArtifact> {
        if request.source_identity.is_empty() || request.source_identity.len() > 4096 {
            return Err(PreparedLowerError::layer_validation(anyhow::anyhow!(
                "source identity is empty or exceeds 4096 bytes"
            )));
        }
        let verified =
            verify_layout(layout, self.limits).map_err(PreparedLowerError::layer_validation)?;
        if request.selected_manifest_digest != verified.manifest_digest {
            return Err(PreparedLowerError::manifest_selection(anyhow::anyhow!(
                "selected manifest identity does not match verified OCI layout"
            )));
        }
        let identities = Identities::new(request).map_err(PreparedLowerError::layer_validation)?;
        let (stage_relative, stage, owned, device, inode, _stage_lock) = {
            let _physical_lock = self.physical_lock()?;
            if let Some(existing) = self
                .available_artifact(identities.artifact_id)
                .map_err(PreparedLowerError::catalog)?
            {
                identities
                    .require_equivalent(&existing)
                    .map_err(PreparedLowerError::catalog)?;
                self.require_artifact_identity(&existing)
                    .map_err(PreparedLowerError::confinement)?;
                return Ok(existing);
            }

            let stage_relative = stage_path(request.operation_id);
            let intent = identities.intent(request.operation_id, &stage_relative);
            self.catalog
                .record_prepare(&intent)
                .map_err(PreparedLowerError::catalog)?;
            let stage = match self.create_stage(&stage_relative) {
                Ok(stage) => stage,
                Err(error) => {
                    let _ = self.catalog.remove_prepare(&intent);
                    return Err(PreparedLowerError::materialization(error));
                }
            };
            let (device, inode) =
                file_identity(&stage).map_err(PreparedLowerError::operating_system)?;
            let stage_lock = match try_exclusive_flock(
                stage
                    .try_clone()
                    .map_err(PreparedLowerError::operating_system)?,
            ) {
                Ok(lock) => lock,
                Err(error) => {
                    if self
                        .remove_exact_path(&stage_relative, device, inode)
                        .is_ok()
                    {
                        let _ = self.catalog.remove_prepare(&intent);
                    }
                    return Err(PreparedLowerError::locking(error));
                }
            };
            let owned = PrepareOperation {
                stage_state: PrepareStageState::Owned,
                device: Some(device),
                inode: Some(inode),
                ..intent.clone()
            };
            if let Err(error) = self.catalog.bind_prepare_stage(&intent, &owned) {
                if self
                    .remove_exact_path(&stage_relative, device, inode)
                    .is_ok()
                {
                    let _ = self.catalog.remove_prepare(&intent);
                }
                return Err(PreparedLowerError::catalog(error));
            }
            (stage_relative, stage, owned, device, inode, stage_lock)
        };
        let materialization_map = if materialize_logical_owners {
            logical_id_map(&request.id_map)
        } else {
            request.id_map.clone()
        };
        if let Err(error) =
            apply_layout(layout, &stage, &verified, &materialization_map, self.limits)
                .and_then(|()| {
                    materialize_destinations(&stage, &request.destinations, &materialization_map)
                })
                .and_then(|()| sync_tree(&stage))
        {
            if let Ok(_physical_lock) = self.physical_lock()
                && self
                    .remove_exact_path(&stage_relative, device, inode)
                    .is_ok()
            {
                let _ = self.catalog.remove_prepare(&owned);
            }
            return Err(PreparedLowerError::materialization(error));
        }
        let physical_lock = self.physical_lock()?;
        self.publish_owned(&owned, &identities, &physical_lock)
            .map_err(PreparedLowerError::publication)
    }

    /// Recovers one exact durable prepare operation without adopting foreign paths.
    ///
    /// # Errors
    ///
    /// Refuses unexpected paths, missing ownership evidence, identity mismatch, and catalog or
    /// filesystem failures.
    pub fn recover_prepare(
        &self,
        operation: &PrepareOperation,
    ) -> PreparedLowerResult<PrepareRecovery> {
        let physical_lock = self.physical_lock()?;
        self.recover_prepare_inner(operation, &physical_lock)
            .map_err(PreparedLowerError::recovery)
    }

    fn recover_prepare_inner(
        &self,
        operation: &PrepareOperation,
        _physical_lock: &ExclusiveFileLock,
    ) -> Result<PrepareRecovery> {
        let expected_stage = stage_path(operation.operation_id);
        if operation.stage_path.as_bytes() != path_bytes(&expected_stage) {
            bail!("prepare operation stage path is not canonical");
        }
        match operation.stage_state {
            PrepareStageState::Intent => {
                if self.open_relative(&expected_stage)?.is_some() {
                    bail!("unowned stage path is present");
                }
                self.catalog.remove_prepare(operation)?;
                Ok(PrepareRecovery::ClearedUnmaterializedIntent)
            }
            PrepareStageState::Owned => {
                let device = operation.device.context("owned prepare lacks device")?;
                let inode = operation.inode.context("owned prepare lacks inode")?;
                if let Some(stage) = self.open_relative(&expected_stage)? {
                    require_identity(&stage, device, inode)?;
                    let _stage_lock = match try_exclusive_flock(stage) {
                        Ok(lock) => lock,
                        Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                            return Ok(PrepareRecovery::PreservedLiveOwnedStage);
                        }
                        Err(error) => return Err(error.into()),
                    };
                    self.remove_exact_path(&expected_stage, device, inode)?;
                    self.catalog.remove_prepare(operation)?;
                    return Ok(PrepareRecovery::RemovedOwnedStage);
                }
                let artifact_relative = artifact_path(operation.expected_artifact_id);
                let Some(artifact_fd) = self.open_relative(&artifact_relative)? else {
                    bail!("owned preparation path is absent without exact recovery evidence");
                };
                if file_identity(&artifact_fd)? != (device, inode) {
                    bail!("published path does not match preparation ownership");
                }
                let artifact =
                    artifact_from_operation(operation, &artifact_relative, device, inode);
                self.publish_catalog_row(operation, &artifact)?;
                Ok(PrepareRecovery::PublishedCompletedStage)
            }
        }
    }

    /// Recovers the complete catalog snapshot of interrupted preparation operations.
    ///
    /// # Errors
    /// Stops at the first foreign, mismatched, unreadable, or changed operation without treating
    /// it as owned absence.
    pub fn recover_preparations(
        &self,
    ) -> PreparedLowerResult<Vec<(PrepareOperationId, PrepareRecovery)>> {
        let physical_lock = self.physical_lock()?;
        self.recover_preparations_inner(&physical_lock)
            .map_err(PreparedLowerError::recovery)
    }

    fn recover_preparations_inner(
        &self,
        physical_lock: &ExclusiveFileLock,
    ) -> Result<Vec<(PrepareOperationId, PrepareRecovery)>> {
        let operations = self.catalog.snapshot()?.prepare_operations;
        operations
            .iter()
            .map(|operation| {
                self.recover_prepare_inner(operation, physical_lock)
                    .map(|outcome| (operation.operation_id, outcome))
            })
            .collect()
    }

    /// Reclaims only the caller-selected exact catalog artifacts.
    ///
    /// # Errors
    /// Refuses an unknown target, duplicate target, any lease, catalog drift, or physical
    /// ownership mismatch. Reclaiming rows resume their existing exact operation.
    pub fn gc_artifacts(
        &self,
        selected: &[PreparedArtifactId],
    ) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
        self.gc_artifacts_inner(selected)
    }

    fn gc_artifacts_inner(
        &self,
        selected: &[PreparedArtifactId],
    ) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
        let mut unique = std::collections::HashSet::new();
        if selected.iter().any(|id| !unique.insert(*id)) {
            return Err(PreparedLowerError::reclaim(anyhow::anyhow!(
                "prepared-artifact GC target is duplicated"
            )));
        }
        let snapshot = self
            .catalog
            .snapshot()
            .map_err(PreparedLowerError::catalog)?;
        let mut results = Vec::with_capacity(selected.len());
        for id in selected {
            let artifact = snapshot
                .prepared_artifacts
                .iter()
                .find(|artifact| artifact.prepared_artifact_id == *id)
                .context("prepared-artifact GC target is absent")
                .map_err(PreparedLowerError::reclaim)?;
            let outcome = match artifact.state {
                ArtifactState::Available => self.reclaim(artifact)?,
                ArtifactState::Reclaiming => self.resume_reclaim(artifact)?,
            };
            results.push(PreparedArtifactGcResult {
                prepared_artifact_id: *id,
                outcome,
            });
        }
        Ok(results)
    }

    /// Moves an exact zero-lease artifact to quarantine, deletes it, and closes reclaim intent.
    ///
    /// # Errors
    ///
    /// Refuses leases, forged catalog rows, physical identity mismatch, foreign paths, and
    /// filesystem or catalog failures.
    pub fn reclaim(&self, expected: &PreparedArtifact) -> PreparedLowerResult<ReclaimOutcome> {
        let reclaiming = self
            .catalog
            .begin_reclaim(expected)
            .map_err(PreparedLowerError::catalog)?;
        self.resume_reclaim(&reclaiming)
    }

    /// Resumes an exact durable reclaim after interruption.
    ///
    /// # Errors
    ///
    /// Refuses non-reclaiming state, path/identity mismatch, missing physical evidence, and
    /// catalog or filesystem failures.
    pub fn resume_reclaim(
        &self,
        reclaiming: &PreparedArtifact,
    ) -> PreparedLowerResult<ReclaimOutcome> {
        let _physical_lock = self.physical_lock()?;
        self.resume_reclaim_inner(reclaiming)
            .map_err(PreparedLowerError::reclaim)
    }

    fn resume_reclaim_inner(&self, reclaiming: &PreparedArtifact) -> Result<ReclaimOutcome> {
        if reclaiming.state != ArtifactState::Reclaiming {
            bail!("artifact is not reclaiming");
        }
        let expected_artifact = artifact_path(reclaiming.prepared_artifact_id);
        if reclaiming.path.as_bytes() != path_bytes(&expected_artifact) {
            bail!("artifact path is not canonical");
        }
        let quarantine = quarantine_path(reclaiming);
        {
            if let Some(canonical) = self.open_relative(&expected_artifact)? {
                if file_identity(&canonical)? == (reclaiming.device, reclaiming.inode) {
                    self.rename_relative_noreplace(&expected_artifact, &quarantine)?;
                    self.sync_fixed_directory("artifacts")?;
                    self.sync_fixed_directory("quarantine")?;
                } else {
                    let quarantined = self
                        .open_relative(&quarantine)?
                        .context("canonical replacement exists without quarantined old artifact")?;
                    require_identity(&quarantined, reclaiming.device, reclaiming.inode)?;
                    let replacement = self
                        .catalog_artifact(reclaiming.prepared_artifact_id)?
                        .context("canonical replacement exists without catalog ownership")?;
                    require_identity(&canonical, replacement.device, replacement.inode)?;
                }
            } else {
                let existing = self
                    .open_relative(&quarantine)?
                    .context("reclaiming artifact and quarantine are both absent")?;
                require_identity(&existing, reclaiming.device, reclaiming.inode)?;
            }
        }
        self.remove_exact_path(&quarantine, reclaiming.device, reclaiming.inode)?;
        match self.catalog.finish_reclaim(reclaiming) {
            Ok(()) => Ok(ReclaimOutcome::Removed),
            Err(CatalogError::OwnershipMismatch {
                row: CatalogRow::PreparedArtifact,
            }) => {
                let replacement = self
                    .catalog_artifact(reclaiming.prepared_artifact_id)?
                    .context("reclaim row changed without an owned replacement")?;
                self.require_artifact_identity(&replacement)?;
                Ok(ReclaimOutcome::ReplacementPreserved)
            }
            Err(error) => Err(error.into()),
        }
    }

    fn publish_owned(
        &self,
        operation: &PrepareOperation,
        identities: &Identities,
        _physical_lock: &ExclusiveFileLock,
    ) -> Result<PreparedArtifact> {
        let stage = stage_path(operation.operation_id);
        let artifact_path = artifact_path(operation.expected_artifact_id);
        if let Some(existing) = self.open_relative(&artifact_path)? {
            if let Some(row) = self.available_artifact(operation.expected_artifact_id)? {
                identities.require_equivalent(&row)?;
                require_identity(&existing, row.device, row.inode)?;
                let device = operation.device.context("owned prepare lacks device")?;
                let inode = operation.inode.context("owned prepare lacks inode")?;
                self.remove_exact_path(&stage, device, inode)?;
                self.catalog.remove_prepare(operation)?;
                return Ok(row);
            }
            let reclaiming = self
                .catalog
                .snapshot()?
                .prepared_artifacts
                .into_iter()
                .find(|candidate| {
                    candidate.prepared_artifact_id == operation.expected_artifact_id
                        && candidate.state == ArtifactState::Reclaiming
                })
                .context("canonical artifact path exists without matching catalog ownership")?;
            require_identity(&existing, reclaiming.device, reclaiming.inode)?;
            let quarantine = quarantine_path(&reclaiming);
            self.rename_relative_noreplace(&artifact_path, &quarantine)?;
            self.sync_fixed_directory("quarantine")?;
        }
        self.rename_relative(&stage, &artifact_path)?;
        self.sync_fixed_directory("artifacts")?;
        let artifact_fd = self
            .open_relative(&artifact_path)?
            .context("published artifact disappeared")?;
        let (device, inode) = file_identity(&artifact_fd)?;
        let artifact = identities.artifact(&artifact_path, device, inode);
        self.publish_catalog_row(operation, &artifact)?;
        Ok(artifact)
    }

    fn publish_catalog_row(
        &self,
        operation: &PrepareOperation,
        artifact: &PreparedArtifact,
    ) -> Result<()> {
        let snapshot = self.catalog.snapshot()?;
        if let Some(reclaiming) = snapshot.prepared_artifacts.into_iter().find(|candidate| {
            candidate.prepared_artifact_id == artifact.prepared_artifact_id
                && candidate.state == ArtifactState::Reclaiming
        }) {
            self.catalog
                .publish_replacement(operation, &reclaiming, artifact)?;
        } else {
            self.catalog.publish(operation, artifact)?;
        }
        Ok(())
    }

    fn available_artifact(&self, id: PreparedArtifactId) -> Result<Option<PreparedArtifact>> {
        Ok(self
            .catalog
            .snapshot()?
            .prepared_artifacts
            .into_iter()
            .find(|artifact| {
                artifact.prepared_artifact_id == id && artifact.state == ArtifactState::Available
            }))
    }

    fn catalog_artifact(&self, id: PreparedArtifactId) -> Result<Option<PreparedArtifact>> {
        Ok(self
            .catalog
            .snapshot()?
            .prepared_artifacts
            .into_iter()
            .find(|artifact| artifact.prepared_artifact_id == id))
    }

    fn require_artifact_identity(&self, artifact: &PreparedArtifact) -> Result<()> {
        let path = PathBuf::from(std::ffi::OsString::from_vec(
            artifact.path.as_bytes().to_vec(),
        ));
        let expected = artifact_path(artifact.prepared_artifact_id);
        if path != expected {
            bail!("artifact path is not canonical");
        }
        let file = self
            .open_relative(&path)?
            .context("artifact path is absent")?;
        require_identity(&file, artifact.device, artifact.inode)
    }

    fn create_stage(&self, relative: &Path) -> Result<File> {
        let stages = open_directory_at(&self.root, Path::new("stages"))?;
        let name = relative.file_name().context("stage has no name")?;
        mkdirat(&stages, name, Mode::from_raw_mode(0o700))?;
        open_directory_at(&stages, Path::new(name))
    }

    fn open_relative(&self, relative: &Path) -> Result<Option<File>> {
        match openat2(
            &self.root,
            relative,
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
            secure_resolve(),
        ) {
            Ok(fd) => Ok(Some(File::from(fd))),
            Err(rustix::io::Errno::NOENT) => Ok(None),
            Err(error) => Err(error.into()),
        }
    }

    fn rename_relative(&self, from: &Path, to: &Path) -> Result<()> {
        renameat(&self.root, from, &self.root, to)
            .with_context(|| format!("renaming {} to {}", from.display(), to.display()))?;
        Ok(())
    }

    fn rename_relative_noreplace(&self, from: &Path, to: &Path) -> Result<()> {
        renameat_with(&self.root, from, &self.root, to, RenameFlags::NOREPLACE).with_context(
            || {
                format!(
                    "renaming {} to {} without replacement",
                    from.display(),
                    to.display()
                )
            },
        )?;
        Ok(())
    }

    fn remove_exact_path(&self, relative: &Path, device: DeviceId, inode: InodeId) -> Result<()> {
        let file = open_directory_at(&self.root, relative)?;
        require_identity(&file, device, inode)?;
        remove_children(&file)?;
        unlinkat(&self.root, relative, AtFlags::REMOVEDIR)?;
        Ok(())
    }

    fn physical_lock(&self) -> PreparedLowerResult<ExclusiveFileLock> {
        let deadline = Instant::now() + PHYSICAL_LOCK_WAIT;
        loop {
            let file = open_lock_at(&self.root).map_err(PreparedLowerError::locking)?;
            match try_exclusive_flock(file) {
                Ok(lock) => return Ok(lock),
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                    let now = Instant::now();
                    if now >= deadline {
                        return Err(PreparedLowerError::locking(anyhow::anyhow!(
                            "prepared-lower physical lock wait expired"
                        )));
                    }
                    thread::sleep(LOCK_RETRY.min(deadline.saturating_duration_since(now)));
                }
                Err(error) => return Err(PreparedLowerError::locking(error)),
            }
        }
    }

    fn sync_fixed_directory(&self, name: &str) -> Result<()> {
        open_directory_at(&self.root, Path::new(name))?.sync_all()?;
        Ok(())
    }
}

fn logical_id_map(target: &IdMap) -> IdMap {
    IdMap {
        uid: target
            .uid
            .iter()
            .map(|range| crate::layer::IdMapRange {
                container_id: range.container_id,
                namespace_id: range.container_id,
                size: range.size,
            })
            .collect(),
        gid: target
            .gid
            .iter()
            .map(|range| crate::layer::IdMapRange {
                container_id: range.container_id,
                namespace_id: range.container_id,
                size: range.size,
            })
            .collect(),
    }
}

fn require_owned_directory(directory: &File) -> Result<()> {
    let stat = rustix::fs::fstat(directory)?;
    if stat.st_uid != rustix::process::getuid().as_raw() || stat.st_mode & 0o7777 != 0o700 {
        bail!("fixed prepared-lower directory must be owned by the invoking user with mode 0700");
    }
    Ok(())
}

struct Identities {
    artifact_id: PreparedArtifactId,
    source: SourceIdentity,
    manifest: SelectedManifestDigest,
    policy: PolicyId,
    id_map: IdMapDigest,
}

impl Identities {
    fn new(request: &PrepareRequest) -> Result<Self> {
        let source = SourceIdentity::new(request.source_identity.clone());
        let manifest = SelectedManifestDigest::new(request.selected_manifest_digest);
        let destinations = canonical_destinations(&request.destinations)?;
        let destination_bytes = destination_policy_bytes(&destinations);
        let policy = PolicyId::new(hash_parts(&[
            b"lazaret-policy-v2",
            POLICY_NAME.as_bytes(),
            &destination_bytes,
        ]));
        let id_map_bytes = canonical_id_map_bytes(&request.id_map)?;
        let id_map = IdMapDigest::new(hash_parts(&[b"lazaret-id-map-v1", &id_map_bytes]));
        let artifact_id = PreparedArtifactId::new(hash_parts(&[
            b"lazaret-prepared-artifact-v1",
            source.as_bytes(),
            manifest.as_bytes(),
            policy.as_bytes(),
            id_map.as_bytes(),
        ]));
        Ok(Self {
            artifact_id,
            source,
            manifest,
            policy,
            id_map,
        })
    }

    fn intent(&self, operation_id: PrepareOperationId, stage: &Path) -> PrepareOperation {
        PrepareOperation {
            operation_id,
            expected_artifact_id: self.artifact_id,
            source_identity: self.source.clone(),
            selected_manifest_digest: self.manifest,
            policy_id: self.policy,
            id_map_digest: self.id_map,
            stage_path: HostPath::new(path_bytes(stage).to_vec()),
            stage_state: PrepareStageState::Intent,
            device: None,
            inode: None,
        }
    }

    fn artifact(&self, path: &Path, device: DeviceId, inode: InodeId) -> PreparedArtifact {
        PreparedArtifact {
            prepared_artifact_id: self.artifact_id,
            source_identity: self.source.clone(),
            selected_manifest_digest: self.manifest,
            policy_id: self.policy,
            id_map_digest: self.id_map,
            path: HostPath::new(path_bytes(path).to_vec()),
            device,
            inode,
            state: ArtifactState::Available,
        }
    }

    fn require_equivalent(&self, artifact: &PreparedArtifact) -> Result<()> {
        if artifact.prepared_artifact_id != self.artifact_id
            || artifact.source_identity != self.source
            || artifact.selected_manifest_digest != self.manifest
            || artifact.policy_id != self.policy
            || artifact.id_map_digest != self.id_map
        {
            bail!("prepared artifact identity inputs differ");
        }
        Ok(())
    }
}

fn canonical_destinations(
    destinations: &[PreparedDestination],
) -> Result<Vec<PreparedDestination>> {
    let mut destinations = destinations.to_vec();
    for destination in &destinations {
        let mut components = destination.path.components();
        if components.next() != Some(std::path::Component::RootDir)
            || components.any(|component| !matches!(component, std::path::Component::Normal(_)))
        {
            bail!("prepared destination is not an absolute normalized path");
        }
    }
    destinations.sort();
    destinations.dedup();
    for pair in destinations.windows(2) {
        if pair[0].path == pair[1].path && pair[0].kind != pair[1].kind {
            bail!("prepared destination has conflicting types");
        }
    }
    Ok(destinations)
}

fn destination_policy_bytes(destinations: &[PreparedDestination]) -> Vec<u8> {
    let mut bytes = b"lazaret-prepared-layout-policy-v1\0".to_vec();
    for destination in destinations {
        bytes.extend_from_slice(destination.path.as_os_str().as_bytes());
        bytes.push(0);
        bytes.push(match destination.kind {
            PreparedDestinationKind::Directory => b'd',
            PreparedDestinationKind::RegularFile => b'f',
        });
        bytes.push(0);
    }
    bytes
}

fn artifact_from_operation(
    operation: &PrepareOperation,
    path: &Path,
    device: DeviceId,
    inode: InodeId,
) -> PreparedArtifact {
    PreparedArtifact {
        prepared_artifact_id: operation.expected_artifact_id,
        source_identity: operation.source_identity.clone(),
        selected_manifest_digest: operation.selected_manifest_digest,
        policy_id: operation.policy_id,
        id_map_digest: operation.id_map_digest,
        path: HostPath::new(path_bytes(path).to_vec()),
        device,
        inode,
        state: ArtifactState::Available,
    }
}

fn file_identity(file: &File) -> Result<(DeviceId, InodeId)> {
    let stat = rustix::fs::fstat(file)?;
    Ok((DeviceId::new(stat.st_dev), InodeId::new(stat.st_ino)))
}

fn require_identity(file: &File, device: DeviceId, inode: InodeId) -> Result<()> {
    if file_identity(file)? == (device, inode) {
        Ok(())
    } else {
        bail!("physical path identity mismatch")
    }
}

fn open_directory_at(parent: impl AsFd, path: &Path) -> Result<File> {
    Ok(File::from(openat2(
        parent,
        path,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
        secure_resolve(),
    )?))
}

fn open_lock_at(root: impl AsFd) -> Result<File> {
    Ok(File::from(openat2(
        root,
        "prepared-lower.lock",
        OFlags::RDWR | OFlags::CREATE | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::from_raw_mode(0o600),
        secure_resolve(),
    )?))
}

fn sync_tree(directory: &File) -> 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?;
            if entry.file_name().to_bytes() != b"." && entry.file_name().to_bytes() != b".." {
                names.push(entry.file_name().to_owned());
            }
        }
        names
    };
    for name in names {
        let path_fd = openat2(
            directory,
            &name,
            OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
            secure_resolve(),
        )?;
        let kind = rustix::fs::FileType::from_raw_mode(rustix::fs::fstat(&path_fd)?.st_mode);
        if kind == rustix::fs::FileType::Directory {
            let child = open_directory_at(
                directory,
                Path::new(std::ffi::OsStr::from_bytes(name.to_bytes())),
            )?;
            sync_tree(&child)?;
        } else if kind == rustix::fs::FileType::RegularFile {
            File::from(openat2(
                directory,
                &name,
                OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
                secure_resolve(),
            )?)
            .sync_all()?;
        }
    }
    directory.sync_all()?;
    Ok(())
}

fn canonical_id_map_bytes(id_map: &IdMap) -> Result<Vec<u8>> {
    let capacity = id_map
        .uid
        .len()
        .checked_add(id_map.gid.len())
        .and_then(|ranges| ranges.checked_mul(12))
        .and_then(|ranges| ranges.checked_add(16))
        .context("ID-map encoding length overflow")?;
    let mut bytes = Vec::with_capacity(capacity);
    for ranges in [&id_map.uid, &id_map.gid] {
        bytes.extend_from_slice(
            &u64::try_from(ranges.len())
                .context("ID-map range count exceeds u64")?
                .to_be_bytes(),
        );
        for range in ranges {
            bytes.extend_from_slice(&range.container_id.to_be_bytes());
            bytes.extend_from_slice(&range.namespace_id.to_be_bytes());
            bytes.extend_from_slice(&range.size.to_be_bytes());
        }
    }
    Ok(bytes)
}

fn hash_parts(parts: &[&[u8]]) -> [u8; 32] {
    let mut hasher = Sha256::new();
    for part in parts {
        hasher.update((part.len() as u64).to_be_bytes());
        hasher.update(part);
    }
    hasher.finalize().into()
}

fn stage_path(id: PrepareOperationId) -> PathBuf {
    Path::new("stages").join(hex(id.as_bytes()))
}

fn artifact_path(id: PreparedArtifactId) -> PathBuf {
    Path::new("artifacts").join(hex(id.as_bytes()))
}

fn quarantine_path(artifact: &PreparedArtifact) -> PathBuf {
    Path::new("quarantine").join(format!(
        "{}-{}-{}",
        hex(artifact.prepared_artifact_id.as_bytes()),
        artifact.device.get(),
        artifact.inode.get()
    ))
}

fn path_bytes(path: &Path) -> &[u8] {
    std::os::unix::ffi::OsStrExt::as_bytes(path.as_os_str())
}

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
        },
    )
}

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

#[cfg(test)]
mod fixed_directory_tests {
    use std::os::unix::fs::PermissionsExt;

    use crate::layer::IdMapRange;

    use super::*;

    #[test]
    fn id_map_identity_encoding_is_explicit_and_fixed() {
        let encoded = canonical_id_map_bytes(&IdMap {
            uid: vec![IdMapRange {
                container_id: 1,
                namespace_id: 2,
                size: 3,
            }],
            gid: vec![IdMapRange {
                container_id: 4,
                namespace_id: 5,
                size: 6,
            }],
        })
        .unwrap();
        assert_eq!(
            encoded,
            [
                0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 1,
                0, 0, 0, 4, 0, 0, 0, 5, 0, 0, 0, 6,
            ]
        );
    }

    #[test]
    fn prepared_lower_root_requires_invoking_owner_and_exact_private_mode() {
        let directory = tempfile::tempdir().unwrap();
        std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o777)).unwrap();
        let root = File::open(directory.path()).unwrap();
        assert!(require_owned_directory(&root).is_err());
        std::fs::set_permissions(directory.path(), std::fs::Permissions::from_mode(0o700)).unwrap();
        assert!(require_owned_directory(&root).is_ok());
    }

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