use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions};
use std::os::fd::AsRawFd;
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::symlink;
use std::os::unix::fs::{FileTypeExt, MetadataExt};
use std::os::unix::net::UnixListener;
use std::path::{Path, PathBuf};

use lazaret_core::configuration::{
    ConfigurationLoadRequest, ResolvedDomainConfiguration, load_trusted_configuration,
};
use lazaret_core::identity::{
    BootId, BundleDigest, ContainerId, CoordinatorId, DeviceId, HostPidLocator, IdMapDigest,
    InodeId, MountId, NetworkSidecarId, PidfsDeviceId, PidfsInodeId, PidfsProcessAuthorityV1,
    PolicyId, PrepareOperationId, PreparedArtifactId, ProcessStartTimeTicks, SandboxId,
    SelectedManifestDigest, SourceIdentity,
};
use lazaret_core::records::{
    ArtifactLease, ArtifactState, DurableDecision, GitRoutingOwnership, HelperOwnership,
    PrepareOperation, PrepareStageState, PreparedArtifact, PreparedLowerOwnership,
    PublicationOwnership, SandboxOwnership, StateAnchorOwnership,
};
use lazaret_linux::filesystem::{current_mount_identity, try_exclusive_flock};
use lazaret_linux::process::{PidfsRecoveryMatch, match_recovery_process};
use lazaret_prepared_lower::PreparedLowerStore;
use lazaret_runtime_supervisor::sandbox::default_sandbox_layout;
use lazaret_state_anchor::StateAnchorLayout;
use lazaret_storage_catalog::Catalog;
use rustix::fs::{AtFlags, unlinkat};

use super::*;
use crate::control::{
    CONTROL_SOCKET_NAME, PUBLICATION_IDENTITY_NAME, STAGED_CONTROL_SOCKET_NAME,
    STAGED_PUBLICATION_IDENTITY_NAME,
};
use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};

fn process(value: u64) -> PidfsProcessIdentityV1 {
    PidfsProcessIdentityV1::new(
        PidfsProcessAuthorityV1 {
            boot_id: BootId::new([1; 16]),
            pidfs_st_dev: PidfsDeviceId::new(value),
            pidfs_st_ino: PidfsInodeId::new(value),
        },
        HostPidLocator::new(u32::try_from(value).unwrap()),
        ProcessStartTimeTicks::new(value),
    )
}

fn generation() -> Generation {
    Generation {
        domain_id: DomainId::new([1; 32]),
        generation_id: GenerationId::new(1),
        configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
        coordinator_id: CoordinatorId::new([3; 32]),
        network_sidecar_id: NetworkSidecarId::new([4; 32]),
        sandbox_id: SandboxId::new([5; 32]),
        durable_decision: DurableDecision::Preparing,
    }
}

#[test]
fn recorded_mount_path_absence_is_successful_recovery_evidence() {
    let tree = tempfile::tempdir().unwrap();
    let absent = tree.path().join("absent");
    let path = HostPath::new(absent.as_os_str().as_bytes().to_vec());
    let mount = MountIdentity {
        mount_id: MountId::new(10),
        device: DeviceId::new(11),
        inode: InodeId::new(12),
    };
    require_mount_absent(Some(&path), Some(mount), "fixture").unwrap();
}

fn participant(role: ParticipantRole, ownership: ParticipantOwnership) -> Participant {
    Participant::snapshot(
        GenerationId::new(1),
        role,
        ParticipantState::Intent,
        Some(ownership),
    )
    .unwrap()
}

#[test]
fn inventory_rejects_default_partial_and_helper_or_fuse_omission() {
    let generation = generation();
    let empty = runtime_recovery_ownership(&[], &generation).unwrap();
    assert!(prove_runtime_owners_absent(&empty).is_err());

    let helper = participant(
        ParticipantRole::Helper,
        ParticipantOwnership::Helper(HelperOwnership {
            network_sidecar_id: None,
            connectivity_helper: PersistedHostProcessOwnership::staged(process(4)),
            runtime_supervisor: PersistedHostProcessOwnership::committed(process(2)),
        }),
    );
    assert!(runtime_recovery_ownership(&[helper], &generation).is_err());

    let incomplete_fuse = participant(
        ParticipantRole::StateAnchor,
        ParticipantOwnership::StateAnchor(StateAnchorOwnership {
            lower_artifact_id: None,
            upper_path: None,
            work_path: None,
            merged_path: Some(HostPath::new(b"/tmp/missing".to_vec())),
            daemon: PersistedHostProcessOwnership::unassigned(),
            user_namespace: Some(NamespaceIdentity {
                device: DeviceId::new(6),
                inode: InodeId::new(7),
            }),
            mount_namespace: Some(NamespaceIdentity {
                device: DeviceId::new(8),
                inode: InodeId::new(9),
            }),
            mount: Some(MountIdentity {
                mount_id: MountId::new(10),
                device: DeviceId::new(11),
                inode: InodeId::new(12),
            }),
        }),
    );
    let prepared = Participant::bare_intent(GenerationId::new(1), ParticipantRole::PreparedLower);
    assert!(runtime_recovery_ownership(&[prepared.clone(), incomplete_fuse], &generation).is_err());

    let staged_state = participant(
        ParticipantRole::StateAnchor,
        ParticipantOwnership::StateAnchor(StateAnchorOwnership {
            lower_artifact_id: None,
            upper_path: None,
            work_path: None,
            merged_path: None,
            daemon: PersistedHostProcessOwnership::staged(process(5)),
            user_namespace: None,
            mount_namespace: None,
            mount: None,
        }),
    );
    let mut opened = 0;
    let partial = runtime_recovery_ownership(&[prepared, staged_state], &generation);
    assert!(partial.is_err());
    if let Ok(partial) = partial {
        let _ = seal_runtime_owners(&partial, |_| {
            opened += 1;
            Err(io::Error::from_raw_os_error(
                rustix::io::Errno::SRCH.raw_os_error(),
            ))
        });
    }
    assert_eq!(opened, 0);
}

#[test]
fn complete_boot_mismatch_inventory_proves_only_after_every_holder_and_mount() {
    let generation = generation();
    let prepared = Participant::bare_intent(GenerationId::new(1), ParticipantRole::PreparedLower);
    let state = participant(
        ParticipantRole::StateAnchor,
        ParticipantOwnership::StateAnchor(StateAnchorOwnership {
            lower_artifact_id: None,
            upper_path: None,
            work_path: None,
            merged_path: None,
            daemon: PersistedHostProcessOwnership::staged(process(5)),
            user_namespace: None,
            mount_namespace: None,
            mount: None,
        }),
    );
    let helper = participant(
        ParticipantRole::Helper,
        ParticipantOwnership::Helper(HelperOwnership {
            network_sidecar_id: None,
            connectivity_helper: PersistedHostProcessOwnership::staged(process(4)),
            runtime_supervisor: PersistedHostProcessOwnership::committed(process(2)),
        }),
    );
    let network = Participant::bare_intent(GenerationId::new(1), ParticipantRole::Network);
    let sandbox = participant(
        ParticipantRole::Sandbox,
        ParticipantOwnership::Sandbox(SandboxOwnership {
            container_id: Some(ContainerId::new([3; 32])),
            runtime_root: Some(HostPath::new(b"/runtime".to_vec())),
            runtime_root_identity: Some(MountIdentity {
                mount_id: MountId::new(13),
                device: DeviceId::new(14),
                inode: InodeId::new(15),
            }),
            bundle_digest: Some(BundleDigest::new([4; 32])),
            host_coordinator: PersistedHostProcessOwnership::committed(process(1)),
            runtime_supervisor: PersistedHostProcessOwnership::staged(process(2)),
            attached_runsc: PersistedHostProcessOwnership::staged(process(3)),
        }),
    );
    let inventory =
        runtime_recovery_ownership(&[prepared, state, helper, network, sandbox], &generation)
            .unwrap();
    assert_eq!(
        prove_runtime_owners_absent(&inventory).unwrap().ownership(),
        &inventory
    );
}

#[test]
fn recovery_preflight_recovers_full_runtime_then_is_idempotent() {
    let fixture = RecoveryFixture::new();
    let mut boundaries = Vec::new();
    fixture
        .recover_with(|boundary, generation| {
            boundaries.push((boundary, generation));
            Ok(())
        })
        .unwrap();
    assert!(
        boundaries
            .iter()
            .any(|(boundary, _)| { *boundary == RecoveryBoundary::StateAnchor })
    );
    assert_eq!(
        fixture.store.snapshot().unwrap().generations[0].durable_decision,
        DurableDecision::Clean
    );
    assert!(
        fixture
            .catalog
            .snapshot()
            .unwrap()
            .artifact_leases
            .is_empty()
    );
    assert!(
        fixture
            .catalog
            .snapshot()
            .unwrap()
            .prepared_artifacts
            .is_empty()
    );
    let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
        fixture
            .configuration
            .runtime_domain_root()
            .canonical_label()
            .as_bytes(),
    ));
    assert!(
        !runtime_domain
            .join("runsc")
            .join(hex_lower(&[6; 32]))
            .exists()
    );

    recover_before_cold_start(
        &fixture.configuration,
        &fixture.store,
        &fixture.catalog,
        &fixture.prepared,
    )
    .unwrap();
}

#[derive(Clone, Copy)]
enum PublicationFixtureBoundary {
    TrulyAbsent,
    AfterStagedIdentity,
    AfterStagedSocket,
}

#[test]
fn bare_publication_intent_requires_every_fixed_label_to_be_absent() {
    let absent = RecoveryFixture::new();
    absent.prepare_bare_publication(PublicationFixtureBoundary::TrulyAbsent);
    recover_publication(&absent.configuration, &absent.store, GenerationId::new(2)).unwrap();
    assert_eq!(absent.publication(2).state(), ParticipantState::Clean);

    let identity = RecoveryFixture::new();
    identity.prepare_bare_publication(PublicationFixtureBoundary::AfterStagedIdentity);
    assert!(
        recover_publication(
            &identity.configuration,
            &identity.store,
            GenerationId::new(2)
        )
        .is_err()
    );
    assert!(
        identity
            .runtime_domain()
            .join(STAGED_PUBLICATION_IDENTITY_NAME)
            .is_file()
    );
    assert_eq!(identity.publication(2).state(), ParticipantState::Intent);
    assert!(identity.publication(2).ownership().is_none());

    let socket = RecoveryFixture::new();
    socket.prepare_bare_publication(PublicationFixtureBoundary::AfterStagedSocket);
    assert!(
        recover_publication(&socket.configuration, &socket.store, GenerationId::new(2)).is_err()
    );
    assert!(
        socket
            .runtime_domain()
            .join(STAGED_PUBLICATION_IDENTITY_NAME)
            .is_file()
    );
    assert!(
        fs::symlink_metadata(socket.runtime_domain().join(STAGED_CONTROL_SOCKET_NAME))
            .unwrap()
            .file_type()
            .is_socket()
    );
    assert_eq!(socket.publication(2).state(), ParticipantState::Intent);
    assert!(socket.publication(2).ownership().is_none());
}

#[test]
fn recovery_preflight_retries_every_recovery_boundary() {
    for failed in [
        RecoveryBoundary::Snapshot,
        RecoveryBoundary::Publication,
        RecoveryBoundary::Sandbox,
        RecoveryBoundary::HelperAndNetwork,
        RecoveryBoundary::GitRouting,
        RecoveryBoundary::StateAnchor,
        RecoveryBoundary::PreparedLower,
        RecoveryBoundary::PrepareOperations,
        RecoveryBoundary::ArtifactGc,
        RecoveryBoundary::FinishGeneration,
    ] {
        let fixture = RecoveryFixture::new();
        let mut injected = false;
        assert!(
            fixture
                .recover_with(|boundary, _| {
                    if boundary == failed && !injected {
                        injected = true;
                        Err("injected recovery boundary".into())
                    } else {
                        Ok(())
                    }
                },)
                .is_err()
        );
        assert!(injected);
        fixture.recover_with(|_, _| Ok(())).unwrap();
        assert_eq!(
            fixture.store.snapshot().unwrap().generations[0].durable_decision,
            DurableDecision::Clean
        );
    }
}

#[test]
fn recovery_preflight_opens_every_exact_owner_once() {
    let fixture = RecoveryFixture::new();
    let mut opened = BTreeMap::new();
    recover_before_cold_start_with_opener(
        &fixture.configuration,
        &fixture.store,
        &fixture.catalog,
        &fixture.prepared,
        |_, _| Ok(()),
        |identity| {
            *opened.entry(identity.host_pid().get()).or_insert(0) += 1;
            Ok(PidfsRecoveryMatch::BootMismatch)
        },
        |root, container| {
            unlinkat(root, container, AtFlags::REMOVEDIR).map_err(|error| error.to_string())
        },
    )
    .unwrap();
    assert_eq!(opened.len(), 6);
    assert!(opened.values().all(|count| *count == 1));
}

#[test]
fn publication_replacement_and_runsc_root_replacement_refuse_untouched() {
    let fixture = RecoveryFixture::new();
    let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
        fixture
            .configuration
            .runtime_domain_root()
            .canonical_label()
            .as_bytes(),
    ));
    let socket = runtime_domain.join(CONTROL_SOCKET_NAME);
    let old_socket = runtime_domain.join("old-control.sock");
    fs::rename(&socket, &old_socket).unwrap();
    let root = File::open(runtime_domain).unwrap();
    drop(
        UnixListener::bind(format!(
            "/proc/self/fd/{}/{}",
            root.as_raw_fd(),
            CONTROL_SOCKET_NAME
        ))
        .unwrap(),
    );
    assert!(fixture.recover_with(|_, _| Ok(())).is_err());
    assert!(socket.exists());

    let fixture = RecoveryFixture::new();
    let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
        fixture
            .configuration
            .runtime_domain_root()
            .canonical_label()
            .as_bytes(),
    ));
    let identity = runtime_domain.join(PUBLICATION_IDENTITY_NAME);
    fs::rename(&identity, runtime_domain.join("old-identity")).unwrap();
    symlink("old-identity", &identity).unwrap();
    assert!(fixture.recover_with(|_, _| Ok(())).is_err());
    assert!(
        fs::symlink_metadata(&identity)
            .unwrap()
            .file_type()
            .is_symlink()
    );

    let fixture = RecoveryFixture::new();
    let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
        fixture
            .configuration
            .runtime_domain_root()
            .canonical_label()
            .as_bytes(),
    ));
    let identity = runtime_domain.join(PUBLICATION_IDENTITY_NAME);
    fs::rename(&identity, runtime_domain.join("old-identity")).unwrap();
    fs::write(&identity, b"replacement").unwrap();
    assert!(fixture.recover_with(|_, _| Ok(())).is_err());
    assert_eq!(fs::read(&identity).unwrap(), b"replacement");

    let fixture = RecoveryFixture::new();
    let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
        fixture
            .configuration
            .runtime_domain_root()
            .canonical_label()
            .as_bytes(),
    ));
    let runsc = runtime_domain.join("runsc");
    let old_runsc = runtime_domain.join("old-runsc");
    fs::rename(&runsc, &old_runsc).unwrap();
    fs::create_dir(&runsc).unwrap();
    let mut deletes = 0;
    let result = recover_before_cold_start_with_opener(
        &fixture.configuration,
        &fixture.store,
        &fixture.catalog,
        &fixture.prepared,
        |_, _| Ok(()),
        |_| Ok(PidfsRecoveryMatch::BootMismatch),
        |_, _| {
            deletes += 1;
            Ok(())
        },
    );
    assert!(result.is_err());
    assert_eq!(deletes, 0);
    assert!(runsc.exists());

    let fixture = RecoveryFixture::new();
    let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
        fixture
            .configuration
            .runtime_domain_root()
            .canonical_label()
            .as_bytes(),
    ));
    let runsc = runtime_domain.join("runsc");
    fs::rename(&runsc, runtime_domain.join("old-runsc")).unwrap();
    symlink("old-runsc", &runsc).unwrap();
    assert!(fixture.recover_with(|_, _| Ok(())).is_err());
    assert!(
        fs::symlink_metadata(&runsc)
            .unwrap()
            .file_type()
            .is_symlink()
    );
}

struct RecoveryFixture {
    _tree: tempfile::TempDir,
    configuration: ResolvedDomainConfiguration,
    store: HostCoordinatorDomainStore,
    catalog: Catalog,
    prepared: PreparedLowerStore,
}

impl RecoveryFixture {
    fn recover_with<Failpoint>(&self, failpoint: Failpoint) -> Result<(), String>
    where
        Failpoint: FnMut(RecoveryBoundary, GenerationId) -> Result<(), String>,
    {
        recover_before_cold_start_with_opener(
            &self.configuration,
            &self.store,
            &self.catalog,
            &self.prepared,
            failpoint,
            match_recovery_process,
            |root, container| {
                unlinkat(root, container, AtFlags::REMOVEDIR).map_err(|error| error.to_string())
            },
        )
    }

    #[allow(clippy::too_many_lines)]
    fn new() -> Self {
        let tree = tempfile::tempdir().unwrap();
        let configuration = test_configuration(tree.path());
        let catalog = Catalog::under_storage_root(configuration.storage_root()).unwrap();
        catalog.initialize().unwrap();
        let prepared =
            PreparedLowerStore::under_storage_root(configuration.storage_root(), catalog.clone())
                .unwrap();
        let artifact_name = hex_lower(&[21; 32]);
        let artifact_path = tree
            .path()
            .join("data/prepared-lower/artifacts")
            .join(&artifact_name);
        fs::create_dir(&artifact_path).unwrap();
        let metadata = artifact_path.metadata().unwrap();
        let artifact = PreparedArtifact {
            prepared_artifact_id: PreparedArtifactId::new([21; 32]),
            source_identity: SourceIdentity::new(vec![22]),
            selected_manifest_digest: SelectedManifestDigest::new([23; 32]),
            policy_id: PolicyId::new([24; 32]),
            id_map_digest: IdMapDigest::new([25; 32]),
            path: HostPath::new(format!("artifacts/{artifact_name}").into_bytes()),
            device: DeviceId::new(metadata.dev()),
            inode: InodeId::new(metadata.ino()),
            state: ArtifactState::Available,
        };
        publish_artifact(&catalog, &artifact);
        let domain = lazaret_core::records::Domain {
            domain_id: configuration.domain_id(),
            profile_name: configuration.selected_profile().to_owned(),
            canonical_workspace: configuration.workspace().canonical_label().clone(),
        };
        let lock = OpenOptions::new()
            .create(true)
            .read(true)
            .truncate(false)
            .write(true)
            .open(tree.path().join("domain.lock"))
            .unwrap();
        let store = HostCoordinatorDomainStore::open(
            try_exclusive_flock(lock).unwrap(),
            &tree.path().join("domain.redb"),
            &domain,
        )
        .unwrap();
        let generation = Generation {
            domain_id: domain.domain_id,
            generation_id: GenerationId::new(1),
            configuration_fingerprint: configuration.configuration_fingerprint(),
            coordinator_id: CoordinatorId::new([3; 32]),
            network_sidecar_id: NetworkSidecarId::new([4; 32]),
            sandbox_id: SandboxId::new([5; 32]),
            durable_decision: DurableDecision::None,
        };
        store
            .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])
            .unwrap();
        let mut generation = generation;
        generation.durable_decision = DurableDecision::Preparing;
        store
            .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])
            .unwrap();
        let layout = StateAnchorLayout::create(&configuration).unwrap();
        let mut anchor = layout.ownership_intent(artifact.prepared_artifact_id);
        let mut staged_anchor = anchor.clone();
        staged_anchor.daemon = PersistedHostProcessOwnership::staged(process(5));
        anchor.daemon = PersistedHostProcessOwnership::committed(process(5));
        anchor.user_namespace = Some(NamespaceIdentity {
            device: DeviceId::new(30),
            inode: InodeId::new(31),
        });
        anchor.mount_namespace = Some(NamespaceIdentity {
            device: DeviceId::new(32),
            inode: InodeId::new(33),
        });
        anchor.mount = Some(MountIdentity {
            mount_id: MountId::new(u64::MAX),
            device: DeviceId::new(34),
            inode: InodeId::new(35),
        });
        let prepared_ownership = PreparedLowerOwnership {
            prepared_artifact_id: Some(artifact.prepared_artifact_id),
            source_identity: Some(artifact.source_identity.clone()),
            lease_state: Some(ParticipantState::LeaseHeld),
            path: Some(artifact.path.clone()),
            device: Some(artifact.device),
            inode: Some(artifact.inode),
            released_state_anchor: None,
        };
        let helper = HelperOwnership {
            network_sidecar_id: None,
            connectivity_helper: PersistedHostProcessOwnership::staged(process(4)),
            runtime_supervisor: PersistedHostProcessOwnership::committed(process(2)),
        };
        let runtime_domain = Path::new(std::ffi::OsStr::from_bytes(
            configuration
                .runtime_domain_root()
                .canonical_label()
                .as_bytes(),
        ));
        let runsc_root = runtime_domain.join("runsc");
        fs::create_dir(&runsc_root).unwrap();
        let container_id = ContainerId::new([6; 32]);
        fs::create_dir(runsc_root.join(hex_lower(container_id.as_bytes()))).unwrap();
        let sandbox = SandboxOwnership {
            container_id: Some(container_id),
            runtime_root: Some(HostPath::new(runsc_root.as_os_str().as_bytes().to_vec())),
            runtime_root_identity: Some(current_mount_identity(&runsc_root).unwrap()),
            bundle_digest: Some(BundleDigest::new([7; 32])),
            host_coordinator: PersistedHostProcessOwnership::committed(process(1)),
            runtime_supervisor: PersistedHostProcessOwnership::staged(process(2)),
            attached_runsc: PersistedHostProcessOwnership::staged(process(3)),
        };
        let publication_socket = runtime_domain.join(CONTROL_SOCKET_NAME);
        let publication_root = File::open(runtime_domain).unwrap();
        drop(
            UnixListener::bind(format!(
                "/proc/self/fd/{}/{}",
                publication_root.as_raw_fd(),
                CONTROL_SOCKET_NAME
            ))
            .unwrap(),
        );
        let publication_identity = runtime_domain.join(PUBLICATION_IDENTITY_NAME);
        fs::write(&publication_identity, b"identity").unwrap();
        let publication = PublicationOwnership {
            staged_socket_path: Some(HostPath::new(
                STAGED_CONTROL_SOCKET_NAME.as_bytes().to_vec(),
            )),
            staged_identity_path: Some(HostPath::new(
                STAGED_PUBLICATION_IDENTITY_NAME.as_bytes().to_vec(),
            )),
            socket_path: Some(HostPath::new(CONTROL_SOCKET_NAME.as_bytes().to_vec())),
            identity_path: Some(HostPath::new(PUBLICATION_IDENTITY_NAME.as_bytes().to_vec())),
            socket_identity: Some(path_identity(runtime_domain, &publication_socket)),
            identity_file_identity: Some(path_identity(runtime_domain, &publication_identity)),
        };
        let git_path = runtime_domain.join("git-routing-fixture");
        fs::create_dir(&git_path).unwrap();
        let git_routing = GitRoutingOwnership {
            mountpoint: Some(HostPath::new(git_path.as_os_str().as_bytes().to_vec())),
            host_root: Some(HostPath::new(b"/git-host".to_vec())),
            isolated_root: Some(HostPath::new(b"/git-isolated".to_vec())),
            daemon: PersistedHostProcessOwnership::staged(process(6)),
            mount: Some(MountIdentity {
                mount_id: MountId::new(u64::MAX - 1),
                device: DeviceId::new(40),
                inode: InodeId::new(41),
            }),
        };
        store
            .apply(&[
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::PreparedLower,
                )),
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::PreparedLower,
                        ParticipantState::Intent,
                        Some(ParticipantOwnership::PreparedLower(
                            PreparedLowerOwnership {
                                lease_state: Some(ParticipantState::Intent),
                                ..prepared_ownership.clone()
                            },
                        )),
                    )
                    .unwrap(),
                ),
            ])
            .unwrap();
        let lease = ArtifactLease {
            prepared_artifact_id: artifact.prepared_artifact_id,
            domain_id: domain.domain_id,
            generation_id: generation.generation_id,
        };
        catalog.acquire_lease_checked(&artifact, &lease).unwrap();
        store
            .apply(&[
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::PreparedLower,
                        ParticipantState::LeaseHeld,
                        Some(ParticipantOwnership::PreparedLower(prepared_ownership)),
                    )
                    .unwrap(),
                ),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::StateAnchor,
                )),
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::StateAnchor,
                        ParticipantState::Intent,
                        Some(ParticipantOwnership::StateAnchor(staged_anchor)),
                    )
                    .unwrap(),
                ),
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::StateAnchor,
                        ParticipantState::Intent,
                        Some(ParticipantOwnership::StateAnchor(anchor.clone())),
                    )
                    .unwrap(),
                ),
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::StateAnchor,
                        ParticipantState::Prepared,
                        Some(ParticipantOwnership::StateAnchor(anchor)),
                    )
                    .unwrap(),
                ),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::Helper,
                )),
                DomainStoreMutation::PutParticipant(participant(
                    ParticipantRole::Helper,
                    ParticipantOwnership::Helper(helper),
                )),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::Network,
                )),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::Sandbox,
                )),
                DomainStoreMutation::PutParticipant(participant(
                    ParticipantRole::Sandbox,
                    ParticipantOwnership::Sandbox(sandbox),
                )),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::GitRouting,
                )),
                DomainStoreMutation::PutParticipant(participant(
                    ParticipantRole::GitRouting,
                    ParticipantOwnership::GitRouting(git_routing),
                )),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation.generation_id,
                    ParticipantRole::Publication,
                )),
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::Publication,
                        ParticipantState::Intent,
                        Some(ParticipantOwnership::Publication(publication.clone())),
                    )
                    .unwrap(),
                ),
                DomainStoreMutation::PutParticipant(
                    Participant::snapshot(
                        generation.generation_id,
                        ParticipantRole::Publication,
                        ParticipantState::Committed,
                        Some(ParticipantOwnership::Publication(publication)),
                    )
                    .unwrap(),
                ),
            ])
            .unwrap();
        Self {
            _tree: tree,
            configuration,
            store,
            catalog,
            prepared,
        }
    }

    fn runtime_domain(&self) -> PathBuf {
        Path::new(std::ffi::OsStr::from_bytes(
            self.configuration
                .runtime_domain_root()
                .canonical_label()
                .as_bytes(),
        ))
        .to_path_buf()
    }

    fn prepare_bare_publication(&self, boundary: PublicationFixtureBoundary) {
        let runtime_domain = self.runtime_domain();
        fs::remove_file(runtime_domain.join(CONTROL_SOCKET_NAME)).unwrap();
        fs::remove_file(runtime_domain.join(PUBLICATION_IDENTITY_NAME)).unwrap();
        let generation = Generation {
            domain_id: self.configuration.domain_id(),
            generation_id: GenerationId::new(2),
            configuration_fingerprint: self.configuration.configuration_fingerprint(),
            coordinator_id: CoordinatorId::new([13; 32]),
            network_sidecar_id: NetworkSidecarId::new([14; 32]),
            sandbox_id: SandboxId::new([15; 32]),
            durable_decision: DurableDecision::None,
        };
        self.store
            .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])
            .unwrap();
        self.store
            .apply(&[
                DomainStoreMutation::PutGeneration(Generation {
                    durable_decision: DurableDecision::Preparing,
                    ..generation
                }),
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    GenerationId::new(2),
                    ParticipantRole::Publication,
                )),
            ])
            .unwrap();
        if !matches!(boundary, PublicationFixtureBoundary::TrulyAbsent) {
            fs::write(
                runtime_domain.join(STAGED_PUBLICATION_IDENTITY_NAME),
                b"identity",
            )
            .unwrap();
        }
        if matches!(boundary, PublicationFixtureBoundary::AfterStagedSocket) {
            let root = File::open(runtime_domain).unwrap();
            drop(
                UnixListener::bind(format!(
                    "/proc/self/fd/{}/{}",
                    root.as_raw_fd(),
                    STAGED_CONTROL_SOCKET_NAME
                ))
                .unwrap(),
            );
        }
    }

    fn publication(&self, generation: u64) -> Participant {
        role(
            &self.store,
            GenerationId::new(generation),
            ParticipantRole::Publication,
        )
        .unwrap()
        .unwrap()
    }
}

fn path_identity(root: &Path, path: &Path) -> MountIdentity {
    let root_identity = current_mount_identity(root).unwrap();
    let metadata = fs::symlink_metadata(path).unwrap();
    MountIdentity {
        mount_id: root_identity.mount_id,
        device: DeviceId::new(metadata.dev()),
        inode: InodeId::new(metadata.ino()),
    }
}

fn publish_artifact(catalog: &Catalog, artifact: &PreparedArtifact) {
    let intent = PrepareOperation {
        operation_id: PrepareOperationId::new([12; 32]),
        expected_artifact_id: artifact.prepared_artifact_id,
        source_identity: artifact.source_identity.clone(),
        selected_manifest_digest: artifact.selected_manifest_digest,
        policy_id: artifact.policy_id,
        id_map_digest: artifact.id_map_digest,
        stage_path: HostPath::new(b"stages/fixture".to_vec()),
        stage_state: PrepareStageState::Intent,
        device: None,
        inode: None,
    };
    catalog.record_prepare(&intent).unwrap();
    let owned = PrepareOperation {
        stage_state: PrepareStageState::Owned,
        device: Some(artifact.device),
        inode: Some(artifact.inode),
        ..intent.clone()
    };
    catalog.bind_prepare_stage(&intent, &owned).unwrap();
    catalog.publish(&owned, artifact).unwrap();
}

fn test_configuration(root: &Path) -> ResolvedDomainConfiguration {
    let configuration_directory = root.join("configuration");
    let workspace = root.join("workspace");
    let runtime = root.join("runtime");
    fs::create_dir(&configuration_directory).unwrap();
    fs::create_dir(&workspace).unwrap();
    fs::create_dir(&runtime).unwrap();
    let configuration_path = configuration_directory.join("lazaret.toml");
    fs::write(
            &configuration_path,
            format!(
                "schema = \"lazaret-trusted-host-config-v1\"\n[storage]\ndata_root = \"{}\"\n[network]\ndefault_allow = [\"192.0.2.10/32\"]\n[profiles.default]\nimage = \"example.invalid/factory:v1\"\n[profiles.default.readiness]\ndenied = [{{ address = \"10.20.30.40\", port = 8443 }}]\npermitted = [{{ address = \"192.0.2.10\", port = 443 }}]\ndns = [{{ resolver = \"198.51.100.53\", query = \"example.invalid\" }}]\n",
                root.join("data").display()
            ),
        )
        .unwrap();
    load_trusted_configuration(ConfigurationLoadRequest {
        explicit_configuration_path: Some(configuration_path),
        selected_profile: "default".into(),
        workspace,
        home: root.join("home"),
        xdg_config_home: None,
        xdg_data_home: None,
        xdg_runtime_dir: runtime,
        environment: BTreeMap::new(),
        sandbox_layout: default_sandbox_layout().unwrap(),
    })
    .unwrap()
}