//! Durable projection recovery planning and rebuild publication.
//!
//! Recovery validates active manifest metadata and referenced artifacts before
//! deciding whether the projection can keep running, needs targeted chunk
//! repair, or must be rebuilt from PostgreSQL source tables by the SQL layer.

use std::collections::BTreeSet;
use std::fs;
use std::path::{Path, PathBuf};

#[cfg(test)]
use crate::persistence::{graph_artifact_checksum_for_path, graph_artifact_version};
use crate::persistence::{
    graph_artifact_edge_type_metadata_for_path, graph_artifact_metadata_for_path,
    projection_manifest_root,
};
use crate::projection::chunk::{
    repair_corrupt_base_chunks, BaseChunkRewriteResult, BaseChunkSource,
};
use crate::projection::edge_type_dictionary::read_manifest_edge_type_dictionary_artifact;
use crate::projection::identity::read_manifest_identity_artifact;
use crate::projection::layered::{ManifestSegmentProvider, SegmentProvider};
use crate::projection::manifest::{
    manifest_file_name, parse_manifest_file_name, ManifestFileRef, ProjectionManifest,
    ProjectionManifestStore,
};
use crate::safety::{GraphError, GraphResult};

/// Recovery action required for the current durable projection artifacts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ProjectionRecoveryAction {
    /// No projection artifacts are present.
    NoProjection,
    /// The active projection manifest and every referenced artifact validate.
    Healthy,
    /// One or more base chunks can be replaced from source table data.
    TargetedChunkRepair,
    /// The projection must be rebuilt from PostgreSQL source tables.
    FullRebuild,
}

/// Recovery inspection result for the active projection artifact root.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ProjectionRecoveryPlan {
    pub(crate) action: ProjectionRecoveryAction,
    pub(crate) generation_id: Option<u64>,
    pub(crate) reason: Option<String>,
}

impl ProjectionRecoveryPlan {
    fn no_projection() -> Self {
        Self {
            action: ProjectionRecoveryAction::NoProjection,
            generation_id: None,
            reason: None,
        }
    }

    fn healthy(manifest: &ProjectionManifest) -> Self {
        Self {
            action: ProjectionRecoveryAction::Healthy,
            generation_id: Some(manifest.generation_id),
            reason: None,
        }
    }

    fn repair(manifest: &ProjectionManifest, reason: impl Into<String>) -> Self {
        Self {
            action: ProjectionRecoveryAction::TargetedChunkRepair,
            generation_id: Some(manifest.generation_id),
            reason: Some(reason.into()),
        }
    }

    fn rebuild(generation_id: Option<u64>, reason: impl Into<String>) -> Self {
        Self {
            action: ProjectionRecoveryAction::FullRebuild,
            generation_id,
            reason: Some(reason.into()),
        }
    }
}

/// Validate the latest active manifest and every referenced segment/chunk.
pub(crate) fn validate_active_projection(root: &Path) -> GraphResult<Option<ProjectionManifest>> {
    let store = ProjectionManifestStore::new(root);
    let _reader_lock = store.acquire_reader_lock()?;
    let Some(manifest) = store.load_latest_current()? else {
        return Ok(None);
    };
    let provider = ManifestSegmentProvider::new(root, &manifest);
    let segments = provider.load_segments()?;
    let base_chunks = provider.load_base_chunks()?;
    let base_edge_type_labels = if manifest.edge_type_dictionary.is_some()
        || !segments.is_empty()
        || !base_chunks.is_empty()
    {
        Some(
            graph_artifact_edge_type_metadata_for_path(&root.join(&manifest.base_artifact_path))?.1,
        )
    } else {
        None
    };
    validate_manifest_edge_type_dictionary(
        root,
        &manifest,
        &segments,
        &base_chunks,
        base_edge_type_labels.as_deref(),
    )?;
    Ok(Some(manifest))
}

/// Decide which recovery action is needed for the current projection root.
pub(crate) fn plan_projection_recovery(root: &Path) -> GraphResult<ProjectionRecoveryPlan> {
    plan_projection_recovery_for_artifact(root, None)
}

/// Decide which recovery action is needed, including base metadata checks.
pub(crate) fn plan_projection_recovery_for_artifact(
    root: &Path,
    graph_path: Option<&Path>,
) -> GraphResult<ProjectionRecoveryPlan> {
    let store = ProjectionManifestStore::new(root);
    let _reader_lock = store.acquire_reader_lock()?;
    let latest_generation = latest_manifest_generation(root)?;
    let manifest = match store.load_latest_current_for_recovery() {
        Ok(Some(manifest)) => manifest,
        Ok(None) => return Ok(ProjectionRecoveryPlan::no_projection()),
        Err(err) => {
            return Ok(ProjectionRecoveryPlan::rebuild(
                latest_generation,
                err.to_string(),
            ));
        }
    };

    let base_metadata_and_labels = if graph_path.is_some() {
        let current_base_path = root.join(&manifest.base_artifact_path);
        match graph_artifact_edge_type_metadata_for_path(&current_base_path) {
            Ok((metadata, labels)) => {
                if let Err(err) =
                    validate_manifest_base_metadata_values(&current_base_path, &manifest, metadata)
                {
                    return Ok(ProjectionRecoveryPlan::rebuild(
                        Some(manifest.generation_id),
                        err.to_string(),
                    ));
                }
                Some((metadata, labels))
            }
            Err(err) => {
                return Ok(ProjectionRecoveryPlan::rebuild(
                    Some(manifest.generation_id),
                    err.to_string(),
                ));
            }
        }
    } else {
        None
    };

    if let Some(reference) = &manifest.relationship_identities {
        let identity_path = root.join(&reference.path);
        if let Err(err) = read_manifest_identity_artifact(
            &identity_path,
            &reference.checksum,
            reference.bytes,
            reference.entry_count,
        ) {
            return Ok(ProjectionRecoveryPlan::rebuild(
                Some(manifest.generation_id),
                err.to_string(),
            ));
        }
    }

    let provider = ManifestSegmentProvider::new(root, &manifest);
    let segments = match provider.load_segments() {
        Ok(segments) => segments,
        Err(err) => {
            return Ok(ProjectionRecoveryPlan::rebuild(
                Some(manifest.generation_id),
                err.to_string(),
            ));
        }
    };
    let base_chunks = match provider.load_base_chunks() {
        Ok(chunks) => chunks,
        Err(err) => return Ok(ProjectionRecoveryPlan::repair(&manifest, err.to_string())),
    };
    let base_edge_type_labels = base_metadata_and_labels
        .as_ref()
        .map(|(_, labels)| labels.as_slice());
    if let Err(err) = validate_manifest_edge_type_dictionary(
        root,
        &manifest,
        &segments,
        &base_chunks,
        base_edge_type_labels,
    ) {
        return Ok(ProjectionRecoveryPlan::rebuild(
            Some(manifest.generation_id),
            err.to_string(),
        ));
    }
    Ok(ProjectionRecoveryPlan::healthy(&manifest))
}

fn validate_manifest_edge_type_dictionary(
    root: &Path,
    manifest: &ProjectionManifest,
    segments: &[crate::projection::segment::DeltaSegment],
    base_chunks: &[crate::projection::segment::DeltaSegment],
    base_edge_type_labels: Option<&[String]>,
) -> GraphResult<()> {
    let governor = crate::resource::load_governor(crate::resource::ByteCount::ZERO);
    validate_manifest_edge_type_dictionary_governed(
        root,
        manifest,
        segments,
        base_chunks,
        base_edge_type_labels,
        &governor,
    )
}

fn validate_manifest_edge_type_dictionary_governed(
    root: &Path,
    manifest: &ProjectionManifest,
    segments: &[crate::projection::segment::DeltaSegment],
    base_chunks: &[crate::projection::segment::DeltaSegment],
    base_edge_type_labels: Option<&[String]>,
    governor: &crate::resource::ResourceGovernor,
) -> GraphResult<()> {
    let count = if let Some(reference) = &manifest.edge_type_dictionary {
        let path = crate::projection::manifest::resolve_manifest_reference(root, &reference.path)?;
        let bound =
            crate::projection::edge_type_dictionary::edge_type_dictionary_decode_upper_bound(
                reference.bytes,
                reference.entry_count,
            )?;
        let _lease = governor
            .reserve_memory(
                crate::resource::ResourcePhase::LoadMetadata,
                crate::resource::ByteCount::from_bytes(bound),
            )
            .map_err(crate::safety::resource_limit_error)?;
        let dictionary = read_manifest_edge_type_dictionary_artifact(
            &path,
            &reference.checksum,
            reference.bytes,
            reference.entry_count,
        )?;
        if base_edge_type_labels.is_some_and(|base| !dictionary.labels().starts_with(base)) {
            return Err(GraphError::CorruptFile {
                reason: "cumulative relationship type dictionary does not preserve the base registry prefix"
                    .to_string(),
            });
        }
        dictionary.labels().len()
    } else if let Some(base) = base_edge_type_labels {
        base.len()
    } else {
        return Ok(());
    };
    for segment in segments.iter().chain(base_chunks) {
        segment.validate_edge_type_ids(count)?;
    }
    Ok(())
}

/// Repair corrupt active base chunks by publishing a replacement generation.
pub(crate) fn repair_active_base_chunks(
    root: &Path,
    source: &impl BaseChunkSource,
) -> GraphResult<Option<BaseChunkRewriteResult>> {
    let Some(manifest) = ProjectionManifestStore::new(root).load_latest_current_for_recovery()?
    else {
        return Ok(None);
    };
    if manifest.base_chunks.is_empty() {
        return Ok(None);
    }
    let result = repair_corrupt_base_chunks(root, &manifest, source)?;
    Ok(Some(result))
}

/// Unpublished generation-specific base planned against one current generation.
///
/// The generated path is confined to the projection root and remains invisible
/// to readers until [`publish_generation_specific_rebuilt_base`] wins the
/// current-generation compare-and-swap.
#[derive(Debug, Clone)]
pub(crate) struct RebuiltBasePublicationPlan {
    root: PathBuf,
    expected_current_generation: Option<u64>,
    predecessor_generation: Option<u64>,
    recovery_publication: bool,
    generation_id: u64,
    candidate_base_name: String,
    candidate_base_path: PathBuf,
    previous: Option<ProjectionManifest>,
}

/// Opaque manifest prepared for one generation-specific rebuilt base.
///
/// Callers may inspect the manifest for candidate loading, but only this
/// module can construct or mutate the value accepted by the publication API.
#[derive(Debug)]
pub(crate) struct PreparedRebuiltBaseManifest {
    manifest: ProjectionManifest,
}

impl PreparedRebuiltBaseManifest {
    /// Borrow the exact manifest that must be validated before publication.
    pub(crate) const fn manifest(&self) -> &ProjectionManifest {
        &self.manifest
    }
}

impl RebuiltBasePublicationPlan {
    /// Return the generation observed before candidate staging began.
    pub(crate) const fn expected_current_generation(&self) -> Option<u64> {
        self.expected_current_generation
    }

    /// Return the generation reserved by this publication plan.
    pub(crate) const fn generation_id(&self) -> u64 {
        self.generation_id
    }

    /// Return the generated root-relative base artifact name.
    pub(crate) fn candidate_base_name(&self) -> &str {
        &self.candidate_base_name
    }

    /// Return the unpublished candidate path under the projection root.
    pub(crate) fn candidate_base_path(&self) -> &Path {
        &self.candidate_base_path
    }

    /// Return whether the current pointer names this plan's generation.
    ///
    /// Callers use this after an ambiguous publication error to avoid deleting
    /// an immutable base that may already be serving readers.
    pub(crate) fn candidate_is_current(&self) -> GraphResult<bool> {
        Ok(
            ProjectionManifestStore::new(&self.root).current_generation_id()?
                == Some(self.generation_id),
        )
    }
}

/// Plan a generation-specific base replacement for a healthy projection root.
///
/// This captures the expected current generation before the caller creates or
/// writes the candidate. The caller must create the returned path with
/// create-new semantics and keep it private until publication succeeds.
///
/// # Errors
///
/// Returns an error when current-generation metadata is corrupt, changes while
/// being captured, the generation counter overflows, or the artifact root
/// cannot be inspected.
pub(crate) fn plan_generation_specific_rebuilt_base(
    root: &Path,
) -> GraphResult<RebuiltBasePublicationPlan> {
    let store = ProjectionManifestStore::new(root);
    let expected_current_generation = store.current_generation_id()?;
    let previous = store.load_latest_metadata()?;
    if previous.as_ref().map(|manifest| manifest.generation_id) != expected_current_generation {
        return Err(GraphError::BuildLocked);
    }
    let generation_id = next_rebuild_generation_id(root)?;
    let candidate_base_name = format!("projection-generation-{generation_id:020}-base.pggraph");
    let candidate_base_path = root.join(&candidate_base_name);
    Ok(RebuiltBasePublicationPlan {
        root: root.to_path_buf(),
        expected_current_generation,
        predecessor_generation: previous.as_ref().map(|manifest| manifest.generation_id),
        recovery_publication: false,
        generation_id,
        candidate_base_name,
        candidate_base_path,
        previous,
    })
}

/// Plan a generation replacement when the current pointer or manifest cannot
/// be decoded safely.
///
/// The generation number remains monotonic across readable manifest filenames,
/// while publication uses the manifest store's raw-pointer recovery switch.
/// No corrupt metadata is trusted for artifact references or timestamps.
pub(crate) fn plan_generation_specific_rebuilt_base_for_recovery(
    root: &Path,
) -> GraphResult<RebuiltBasePublicationPlan> {
    let predecessor_generation = latest_manifest_generation(root)?;
    let generation_id = next_rebuild_generation_id(root)?;
    let candidate_base_name = format!("projection-generation-{generation_id:020}-base.pggraph");
    let candidate_base_path = root.join(&candidate_base_name);
    Ok(RebuiltBasePublicationPlan {
        root: root.to_path_buf(),
        expected_current_generation: None,
        predecessor_generation,
        recovery_publication: true,
        generation_id,
        candidate_base_name,
        candidate_base_path,
        previous: None,
    })
}

/// Prepare the manifest for a staged generation-specific base.
///
/// The candidate must already be complete and fsynced. This function computes
/// its persisted checksum,
/// creates the base-only manifest, inherits operation timestamps, and records
/// all superseded artifacts without changing the current pointer. The caller
/// can therefore pass the returned manifest to the production candidate loader
/// before invoking [`publish_prepared_generation_specific_rebuilt_base`].
///
/// # Errors
///
/// Returns validation or I/O errors when the candidate is missing or invalid,
/// its checksum cannot be read, or manifest metadata cannot be created.
pub(crate) fn prepare_generation_specific_rebuilt_base_manifest(
    plan: &RebuiltBasePublicationPlan,
    sync_watermark: i64,
) -> GraphResult<PreparedRebuiltBaseManifest> {
    if !plan.candidate_base_path.is_file() {
        return Err(GraphError::Internal(format!(
            "rebuilt base candidate does not exist: {}",
            plan.candidate_base_path.display()
        )));
    }
    let candidate_metadata = graph_artifact_metadata_for_path(&plan.candidate_base_path)?;
    let mut manifest = ProjectionManifest::base_only(
        plan.generation_id,
        plan.candidate_base_name.clone(),
        candidate_metadata.checksum(),
        candidate_metadata.version,
        sync_watermark,
        now_unix_micros()?,
    );
    manifest.previous_generation_id = plan.predecessor_generation;
    if let Some(previous) = plan.previous.as_ref() {
        manifest.inherit_operation_timestamps(previous);
        manifest.obsolete_files = previous.obsolete_files.clone();
        append_superseded_projection_files(
            &plan.root,
            previous,
            &manifest.base_artifact_path,
            &mut manifest.obsolete_files,
        );
    }
    Ok(PreparedRebuiltBaseManifest { manifest })
}

/// Publish a prepared rebuilt-base manifest through normal manifest CAS.
///
/// The manifest must be the exact candidate described by `plan`. Its
/// generation, base name, artifact version, checksum, predecessor, and current
/// generation expectation are revalidated immediately before publication.
///
/// # Errors
///
/// Returns [`GraphError::BuildLocked`] when another publisher changed the
/// current generation. Returns [`GraphError::CorruptFile`] when the prepared
/// manifest does not match the plan or the staged candidate bytes. Returns
/// durable manifest publication errors from
/// [`ProjectionManifestStore::publish_if_current`].
pub(crate) fn publish_prepared_generation_specific_rebuilt_base(
    plan: &RebuiltBasePublicationPlan,
    prepared: &PreparedRebuiltBaseManifest,
) -> GraphResult<()> {
    let manifest = prepared.manifest();
    let store = ProjectionManifestStore::new(&plan.root);
    if !plan.recovery_publication
        && store.current_generation_id()? != plan.expected_current_generation
    {
        return Err(GraphError::BuildLocked);
    }
    let candidate_metadata = graph_artifact_metadata_for_path(&plan.candidate_base_path)?;
    if manifest.generation_id != plan.generation_id
        || manifest.base_artifact_path != plan.candidate_base_name
        || manifest.base_artifact_version != candidate_metadata.version
        || manifest.previous_generation_id != plan.predecessor_generation
    {
        return Err(GraphError::CorruptFile {
            reason: "prepared rebuilt-base manifest does not match its publication plan".into(),
        });
    }
    if manifest.base_artifact_checksum != candidate_metadata.checksum() {
        return Err(GraphError::CorruptFile {
            reason: "prepared rebuilt-base manifest checksum does not match its candidate".into(),
        });
    }
    if plan.recovery_publication {
        store.publish_for_recovery(manifest)?;
    } else {
        store.publish_if_current(manifest, plan.expected_current_generation)?;
    }
    Ok(())
}

/// Prepare and publish a staged generation-specific base through normal CAS.
///
/// Callers that must validate the candidate through the production loader
/// before publication should use
/// [`prepare_generation_specific_rebuilt_base_manifest`] followed by
/// [`publish_prepared_generation_specific_rebuilt_base`].
///
/// # Errors
///
/// Returns the preparation and CAS publication errors documented by those two
/// functions.
pub(crate) fn publish_generation_specific_rebuilt_base(
    plan: &RebuiltBasePublicationPlan,
    sync_watermark: i64,
) -> GraphResult<ProjectionManifest> {
    let prepared = prepare_generation_specific_rebuilt_base_manifest(plan, sync_watermark)?;
    publish_prepared_generation_specific_rebuilt_base(plan, &prepared)?;
    Ok(prepared.manifest)
}

/// Publish a fresh base-only manifest after a successful PostgreSQL rebuild.
pub(crate) fn publish_rebuilt_base_manifest(
    graph_path: &Path,
    sync_watermark: i64,
) -> GraphResult<ProjectionManifest> {
    let root = projection_manifest_root(graph_path);
    let store = ProjectionManifestStore::new(&root);
    let previous_generation_id = latest_manifest_generation(&root)?;
    let previous = match store.load_latest_metadata() {
        Ok(previous) => previous,
        Err(GraphError::CorruptFile { .. } | GraphError::IncompatibleVersion(_)) => None,
        Err(err) => return Err(err),
    };
    let generation_id = next_rebuild_generation_id(&root)?;
    let base_artifact_path = graph_path
        .file_name()
        .and_then(|value| value.to_str())
        .ok_or_else(|| GraphError::Internal("graph artifact path has no file name".into()))?;
    let artifact_metadata = graph_artifact_metadata_for_path(graph_path)?;
    let mut manifest = ProjectionManifest::base_only(
        generation_id,
        base_artifact_path,
        artifact_metadata.checksum(),
        artifact_metadata.version,
        sync_watermark,
        now_unix_micros()?,
    );
    if let Some(previous) = previous.as_ref() {
        manifest.previous_generation_id = Some(previous.generation_id);
        manifest.inherit_operation_timestamps(previous);
        manifest.obsolete_files = previous.obsolete_files.clone();
        append_superseded_projection_files(
            &root,
            previous,
            &manifest.base_artifact_path,
            &mut manifest.obsolete_files,
        );
    } else {
        manifest.previous_generation_id = previous_generation_id;
    }
    store.publish_for_recovery(&manifest)?;
    Ok(manifest)
}

fn append_superseded_projection_files(
    root: &Path,
    previous: &ProjectionManifest,
    replacement_base_path: &str,
    obsolete: &mut Vec<ManifestFileRef>,
) {
    let mut paths = obsolete
        .iter()
        .map(|reference| reference.path.clone())
        .collect::<std::collections::HashSet<_>>();
    let mut push = |path: &str, known_bytes: Option<u64>| {
        if !paths.insert(path.to_string()) {
            return;
        }
        let bytes = known_bytes.unwrap_or_else(|| {
            std::fs::metadata(root.join(path))
                .map(|metadata| metadata.len())
                .unwrap_or(0)
        });
        obsolete.push(ManifestFileRef {
            path: path.to_string(),
            bytes,
        });
    };
    for segment in &previous.segments {
        push(&segment.path, None);
    }
    if let Some(identity) = previous.relationship_identities.as_ref() {
        push(&identity.path, Some(identity.bytes));
    }
    if let Some(dictionary) = previous.edge_type_dictionary.as_ref() {
        push(&dictionary.path, Some(dictionary.bytes));
    }
    for chunk in &previous.base_chunks {
        push(&chunk.path, None);
    }
    if previous.base_artifact_path != replacement_base_path {
        push(&previous.base_artifact_path, None);
        for sidecar in [".sync", ".projection_mode"] {
            push(&format!("{}{sidecar}", previous.base_artifact_path), None);
        }
    }
}

fn validate_manifest_base_metadata(
    graph_path: &Path,
    manifest: &ProjectionManifest,
) -> GraphResult<()> {
    let artifact_metadata = graph_artifact_metadata_for_path(graph_path)?;
    validate_manifest_base_metadata_values(graph_path, manifest, artifact_metadata)
}

fn validate_manifest_base_metadata_values(
    graph_path: &Path,
    manifest: &ProjectionManifest,
    artifact_metadata: crate::persistence::GraphArtifactMetadata,
) -> GraphResult<()> {
    let expected_base = graph_path
        .file_name()
        .and_then(|value| value.to_str())
        .ok_or_else(|| GraphError::Internal("graph artifact path has no file name".into()))?;
    if manifest.base_artifact_path != expected_base {
        return Err(GraphError::CorruptFile {
            reason: format!(
                "projection manifest: base artifact '{}' does not match loaded artifact '{}'",
                manifest.base_artifact_path, expected_base
            ),
        });
    }
    if manifest.base_artifact_version != artifact_metadata.version {
        return Err(GraphError::IncompatibleVersion(format!(
            "projection manifest references base artifact version {}; expected {}",
            manifest.base_artifact_version, artifact_metadata.version
        )));
    }
    let expected_checksum = artifact_metadata.checksum();
    if manifest.base_artifact_checksum != expected_checksum {
        return Err(GraphError::CorruptFile {
            reason: format!(
                "projection manifest: base artifact checksum '{}' does not match loaded artifact checksum '{}'",
                manifest.base_artifact_checksum, expected_checksum
            ),
        });
    }
    Ok(())
}

/// Return the next generation id after every reserved generation in `root`.
///
/// A generation-specific base or sidecar reserves its identifier even when a
/// crash occurs before the manifest is created. Skipping that identifier keeps
/// the next candidate compatible with create-new publication semantics.
pub(crate) fn next_rebuild_generation_id(root: &Path) -> GraphResult<u64> {
    latest_known_generation(root)?
        .unwrap_or(0)
        .checked_add(1)
        .ok_or_else(|| GraphError::Internal("projection generation id overflowed".into()))
}

/// Return whether the artifact root contains a published or quarantined
/// projection generation that a persisted rebuild must supersede.
pub(crate) fn has_projection_generation(root: &Path) -> GraphResult<bool> {
    latest_known_generation(root).map(|generation| generation.is_some())
}

fn latest_known_generation(root: &Path) -> GraphResult<Option<u64>> {
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(err) => {
            return Err(GraphError::Internal(format!(
                "projection recovery read artifact directory failed for {}: {err}",
                root.display()
            )));
        }
    };
    let mut latest = None;
    for entry in entries {
        let entry = entry.map_err(|err| {
            GraphError::Internal(format!(
                "projection recovery read artifact entry failed for {}: {err}",
                root.display()
            ))
        })?;
        if !entry
            .file_type()
            .map_err(|err| {
                GraphError::Internal(format!(
                    "projection recovery read artifact file type failed for {}: {err}",
                    entry.path().display()
                ))
            })?
            .is_file()
        {
            continue;
        }
        let Some(file_name) = entry.file_name().to_str().map(str::to_owned) else {
            continue;
        };
        let generation_id = parse_manifest_file_name(&file_name)
            .or_else(|| parse_candidate_artifact_generation(&file_name))
            .or_else(|| {
                file_name
                    .split_once(".invalid-")
                    .and_then(|(original, _)| parse_manifest_file_name(original))
            });
        if let Some(generation_id) = generation_id {
            if latest.is_none_or(|current| generation_id > current) {
                latest = Some(generation_id);
            }
        }
    }
    Ok(latest)
}

fn parse_candidate_artifact_generation(file_name: &str) -> Option<u64> {
    let generation = if let Some(suffix) = file_name.strip_prefix("projection-generation-") {
        let (generation, remainder) = suffix.split_at_checked(20)?;
        if !remainder.starts_with('-')
            && !remainder.starts_with(".json")
            && !remainder.starts_with(".tmp-")
        {
            return None;
        }
        generation
    } else if let Some(suffix) = file_name.strip_prefix("relationship-identities-") {
        let (generation, remainder) = suffix.split_at_checked(20)?;
        if remainder != ".bin" {
            return None;
        }
        generation
    } else if let Some(suffix) = file_name.strip_prefix("relationship-types-") {
        let (generation, remainder) = suffix.split_at_checked(20)?;
        if remainder != ".bin" {
            return None;
        }
        generation
    } else {
        return None;
    };
    if !generation.bytes().all(|byte| byte.is_ascii_digit()) {
        return None;
    }
    generation.parse().ok()
}

/// Remove one generation candidate proven unpublished by an unchanged pointer.
///
/// This is the cancellation recovery path for work that crossed PostgreSQL's
/// error boundary before ordinary Rust cleanup could run. The caller must hold
/// the graph's PostgreSQL build advisory lock so another publisher cannot be
/// constructing the same reusable candidate generation. The publication lock
/// makes the current-generation comparison and cleanup one atomic decision.
/// Files referenced by every other manifest are always protected; only the
/// recorded unpublished generation and its private artifacts are eligible for
/// removal.
pub(crate) fn cleanup_interrupted_replacement(
    root: &Path,
    expected_current: Option<u64>,
    candidate_generation: Option<u64>,
) -> GraphResult<usize> {
    let store = ProjectionManifestStore::new(root);
    let _publication_lock = store.acquire_publication_lock()?;
    if store.current_generation_id_for_recovery()? != expected_current {
        return Ok(0);
    }

    let Some(candidate_generation) = candidate_generation else {
        return Ok(0);
    };
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(err) => {
            return Err(GraphError::Internal(format!(
                "projection replacement cleanup read artifact directory failed for {}: {err}",
                root.display()
            )));
        }
    };
    let mut manifests = Vec::new();
    let mut candidate_artifacts = Vec::new();
    for entry in entries {
        let entry = entry.map_err(|err| {
            GraphError::Internal(format!(
                "projection replacement cleanup read artifact entry failed for {}: {err}",
                root.display()
            ))
        })?;
        if !entry
            .file_type()
            .map_err(|err| {
                GraphError::Internal(format!(
                    "projection replacement cleanup read file type failed for {}: {err}",
                    entry.path().display()
                ))
            })?
            .is_file()
        {
            continue;
        }
        let Some(file_name) = entry.file_name().to_str().map(str::to_owned) else {
            continue;
        };
        if let Some(generation_id) = parse_manifest_file_name(&file_name) {
            let manifest = fs::read_to_string(entry.path())
                .ok()
                .and_then(|raw| ProjectionManifest::from_json(&raw).ok())
                .filter(|manifest| manifest.generation_id == generation_id);
            manifests.push((generation_id, entry.path(), manifest));
        } else if let Some(generation_id) = parse_candidate_artifact_generation(&file_name) {
            candidate_artifacts.push((generation_id, entry.path()));
        }
    }

    let mut protected = BTreeSet::new();
    for (generation_id, _, manifest) in &manifests {
        if *generation_id != candidate_generation {
            if let Some(manifest) = manifest {
                insert_active_manifest_references(root, manifest, &mut protected)?;
            }
        }
    }

    let mut candidates = BTreeSet::new();
    for (generation_id, manifest_path, manifest) in &manifests {
        if *generation_id != candidate_generation {
            continue;
        }
        candidates.insert(manifest_path.clone());
        if let Some(manifest) = manifest {
            insert_active_manifest_references(root, manifest, &mut candidates)?;
        }
    }
    candidates.extend(
        candidate_artifacts
            .into_iter()
            .filter(|(generation_id, _)| *generation_id == candidate_generation)
            .map(|(_, path)| path),
    );

    let mut removed = 0usize;
    for candidate in candidates.difference(&protected) {
        match fs::remove_file(candidate) {
            Ok(()) => removed = removed.saturating_add(1),
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
            Err(err) => {
                return Err(GraphError::Internal(format!(
                    "projection replacement cleanup failed for {}: {err}",
                    candidate.display()
                )));
            }
        }
    }
    Ok(removed)
}

fn insert_active_manifest_references(
    root: &Path,
    manifest: &ProjectionManifest,
    paths: &mut BTreeSet<PathBuf>,
) -> GraphResult<()> {
    let base = crate::projection::manifest::resolve_manifest_reference(
        root,
        &manifest.base_artifact_path,
    )?;
    paths.insert(base.clone());
    for suffix in [".sync", ".projection_mode"] {
        let mut sidecar = base.as_os_str().to_os_string();
        sidecar.push(suffix);
        paths.insert(PathBuf::from(sidecar));
    }
    if let Some(identities) = &manifest.relationship_identities {
        paths.insert(crate::projection::manifest::resolve_manifest_reference(
            root,
            &identities.path,
        )?);
    }
    if let Some(dictionary) = &manifest.edge_type_dictionary {
        paths.insert(crate::projection::manifest::resolve_manifest_reference(
            root,
            &dictionary.path,
        )?);
    }
    for segment in &manifest.segments {
        paths.insert(crate::projection::manifest::resolve_manifest_reference(
            root,
            &segment.path,
        )?);
    }
    for chunk in &manifest.base_chunks {
        paths.insert(crate::projection::manifest::resolve_manifest_reference(
            root,
            &chunk.path,
        )?);
    }
    Ok(())
}

/// Move the latest final manifest aside so a full rebuild can reload safely.
pub(crate) fn quarantine_latest_manifest(root: &Path) -> GraphResult<Option<PathBuf>> {
    let Some(generation_id) = latest_manifest_generation(root)? else {
        return Ok(None);
    };
    let path = root.join(manifest_file_name(generation_id));
    if !path.is_file() {
        return Ok(None);
    }
    for attempt in 0..128 {
        let quarantine = root.join(format!(
            "{}.invalid-{attempt}",
            manifest_file_name(generation_id)
        ));
        match fs::rename(&path, &quarantine) {
            Ok(()) => return Ok(Some(quarantine)),
            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(err) => {
                return Err(GraphError::Internal(format!(
                    "projection recovery quarantine failed for {}: {err}",
                    path.display()
                )));
            }
        }
    }
    Err(GraphError::Internal(
        "projection recovery quarantine path kept colliding".into(),
    ))
}

/// Restore a manifest previously moved by [`quarantine_latest_manifest`].
pub(crate) fn restore_quarantined_manifest(quarantine_path: &Path) -> GraphResult<()> {
    if !quarantine_path.exists() {
        return Ok(());
    }
    let file_name = quarantine_path
        .file_name()
        .and_then(|value| value.to_str())
        .ok_or_else(|| {
            GraphError::Internal("projection quarantine path has no file name".into())
        })?;
    let Some(original_name) = file_name.split(".invalid-").next() else {
        return Err(GraphError::Internal(format!(
            "projection quarantine path has invalid name: {}",
            quarantine_path.display()
        )));
    };
    if original_name == file_name {
        return Err(GraphError::Internal(format!(
            "projection quarantine path has no invalid suffix: {}",
            quarantine_path.display()
        )));
    }
    let original_path = quarantine_path.with_file_name(original_name);
    fs::rename(quarantine_path, &original_path).map_err(|err| {
        GraphError::Internal(format!(
            "projection recovery restore failed for {}: {err}",
            quarantine_path.display()
        ))
    })
}

fn latest_manifest_generation(root: &Path) -> GraphResult<Option<u64>> {
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(err) => {
            return Err(GraphError::Internal(format!(
                "projection recovery read artifact directory failed for {}: {err}",
                root.display()
            )));
        }
    };
    let mut latest = None;
    for entry in entries {
        let entry = entry.map_err(|err| {
            GraphError::Internal(format!(
                "projection recovery read artifact entry failed for {}: {err}",
                root.display()
            ))
        })?;
        if !entry
            .file_type()
            .map_err(|err| {
                GraphError::Internal(format!(
                    "projection recovery read artifact file type failed for {}: {err}",
                    entry.path().display()
                ))
            })?
            .is_file()
        {
            continue;
        }
        let Some(file_name) = entry.file_name().to_str().map(str::to_owned) else {
            continue;
        };
        let Some(generation_id) = parse_manifest_file_name(&file_name) else {
            continue;
        };
        if latest.is_none_or(|current| generation_id > current) {
            latest = Some(generation_id);
        }
    }
    Ok(latest)
}

fn now_unix_micros() -> GraphResult<i64> {
    let duration = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_err(|err| GraphError::Internal(format!("system clock before Unix epoch: {err}")))?;
    i64::try_from(duration.as_micros())
        .map_err(|_| GraphError::Internal("current timestamp exceeds i64 micros".into()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::projection::chunk::EdgeStoreChunkSource;
    use crate::projection::edge_type_dictionary::{
        write_edge_type_dictionary_artifact, EdgeTypeDictionary,
    };
    use crate::projection::identity::{write_identity_artifact, RelationshipIdentityDictionary};
    use crate::projection::manifest::{
        ManifestChunkRef, ManifestEdgeTypeDictionaryRef, ManifestIdentityRef, ManifestSegmentRef,
    };
    use crate::projection::segment::{DeltaSegment, SegmentEdge, SegmentKind};
    use crate::projection::test_fixtures::{edge_store_from_tuples, ProjectionArtifactDir};
    use crate::types::TraversalDirection;

    fn test_governor() -> crate::resource::ResourceGovernor {
        crate::resource::ResourceGovernor::new(crate::resource::ResourceLimits::memory_only(
            crate::resource::MemoryBudget::new(crate::resource::ByteCount::from_bytes(u64::MAX)),
        ))
    }

    #[test]
    fn interrupted_replacement_cleanup_removes_only_newer_unpublished_files() {
        let dir = ProjectionArtifactDir::new(
            "interrupted_replacement_cleanup_removes_only_newer_unpublished_files",
        );
        let current_base = dir.path().join("base.pggraph");
        write_file(current_base.clone(), b"current");
        let shared_segment = dir.path().join("shared.pggraph-delta");
        write_file(shared_segment.clone(), b"shared");
        let mut current = base_manifest(1);
        current
            .segments
            .push(segment_ref(dir.path(), &shared_segment, "crc32:shared"));
        let store = ProjectionManifestStore::new(dir.path());
        store.publish(&current).expect("current publishes");

        let candidate_base = dir
            .path()
            .join("projection-generation-00000000000000000002-base.pggraph");
        write_file(candidate_base.clone(), b"candidate");
        let candidate_sync = dir
            .path()
            .join("projection-generation-00000000000000000002-base.pggraph.sync");
        write_file(candidate_sync.clone(), b"1");
        let candidate_segment = dir
            .path()
            .join("projection-generation-00000000000000000002-segment-00000000.pggraph-delta");
        write_file(candidate_segment.clone(), b"segment");
        let candidate_chunk = dir
            .path()
            .join("projection-generation-00000000000000000002-base-chunk-00000000.pggraph-chunk");
        write_file(candidate_chunk.clone(), b"chunk");
        let candidate_temp = dir
            .path()
            .join("projection-generation-00000000000000000002.tmp-cancelled");
        write_file(candidate_temp.clone(), b"partial manifest");
        let candidate_identities = dir
            .path()
            .join("relationship-identities-00000000000000000002.bin");
        write_file(candidate_identities.clone(), b"identities");
        let mut candidate = ProjectionManifest::base_only(
            2,
            candidate_base
                .file_name()
                .and_then(|name| name.to_str())
                .expect("candidate name is UTF-8"),
            "crc32:candidate",
            graph_artifact_version(),
            1,
            2,
        );
        candidate.segments = current.segments.clone();
        let candidate_manifest = store.manifest_path(2);
        fs::write(
            &candidate_manifest,
            candidate.to_pretty_json().expect("candidate encodes"),
        )
        .expect("candidate manifest writes");

        let removed = cleanup_interrupted_replacement(dir.path(), Some(1), Some(2))
            .expect("interrupted candidate cleans up");

        assert_eq!(removed, 7);
        assert!(current_base.exists());
        assert!(shared_segment.exists());
        assert!(!candidate_base.exists());
        assert!(!candidate_sync.exists());
        assert!(!candidate_segment.exists());
        assert!(!candidate_chunk.exists());
        assert!(!candidate_temp.exists());
        assert!(!candidate_identities.exists());
        assert!(!candidate_manifest.exists());
        assert_eq!(
            store.current_generation_id().expect("current reads"),
            Some(1)
        );
    }

    #[test]
    fn interrupted_replacement_cleanup_preserves_a_candidate_that_won_publication() {
        let dir = ProjectionArtifactDir::new(
            "interrupted_replacement_cleanup_preserves_a_candidate_that_won_publication",
        );
        write_file(dir.path().join("base.pggraph"), b"current");
        let store = ProjectionManifestStore::new(dir.path());
        store.publish(&base_manifest(1)).expect("current publishes");
        let next_base = dir.path().join("next.pggraph");
        write_file(next_base.clone(), b"next");
        let next = ProjectionManifest::base_only(
            2,
            "next.pggraph",
            "crc32:next",
            graph_artifact_version(),
            1,
            2,
        );
        store
            .publish_if_current(&next, Some(1))
            .expect("next publishes");

        let removed = cleanup_interrupted_replacement(dir.path(), Some(1), Some(2))
            .expect("published candidate is inspected");

        assert_eq!(removed, 0);
        assert!(next_base.exists());
        assert!(store.manifest_path(2).exists());
        assert_eq!(
            store.current_generation_id().expect("current reads"),
            Some(2)
        );
    }

    #[test]
    fn load_corrupt_active_segment_repairs_or_rebuilds() {
        let dir = ProjectionArtifactDir::new("load_corrupt_active_segment_repairs_or_rebuilds");
        write_file(dir.path().join("base.pggraph"), b"base");
        let segment_path = dir.path().join("active.pggraph-delta");
        let segment = edge_segment(1, 0, &[(0, 1, 1)]);
        segment
            .write_to_path(&segment_path)
            .expect("segment writes");
        let mut manifest = base_manifest(1);
        manifest
            .segments
            .push(segment_ref(dir.path(), &segment_path, "crc32:00000000"));
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");

        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");

        assert_eq!(plan.action, ProjectionRecoveryAction::FullRebuild);
        assert_eq!(plan.generation_id, Some(1));
    }

    #[test]
    fn corrupt_relationship_identity_artifact_requires_full_rebuild() {
        let dir = ProjectionArtifactDir::new(
            "corrupt_relationship_identity_artifact_requires_full_rebuild",
        );
        write_file(dir.path().join("base.pggraph"), b"base");
        let identity_path = dir
            .path()
            .join("relationship-identities-00000000000000000001.bin");
        let dictionary = RelationshipIdentityDictionary::try_from_identities(vec![None])
            .expect("empty identity dictionary validates");
        let (checksum, bytes) = write_identity_artifact(dir.path(), &identity_path, &dictionary)
            .expect("identity artifact writes");
        let mut manifest = base_manifest(1);
        manifest.relationship_identities = Some(ManifestIdentityRef {
            path: identity_path
                .file_name()
                .and_then(|name| name.to_str())
                .expect("identity path is UTF-8")
                .to_string(),
            checksum,
            bytes,
            entry_count: 1,
        });
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        write_file(&identity_path, b"corrupt");

        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");

        assert_eq!(plan.action, ProjectionRecoveryAction::FullRebuild);
        assert_eq!(plan.generation_id, Some(1));
        assert!(plan
            .reason
            .as_deref()
            .is_some_and(|reason| reason.contains("relationship identity")));
    }

    #[test]
    fn recovery_rejects_missing_or_corrupt_edge_type_dictionary() {
        let dir = ProjectionArtifactDir::new("recovery_rejects_missing_edge_type_dictionary");
        write_file(dir.path().join("base.pggraph"), b"base");
        let mut manifest = base_manifest(1);
        let dictionary_path = dir
            .path()
            .join("relationship-types-00000000000000000001.bin");
        write_file(&dictionary_path, b"candidate");
        manifest.edge_type_dictionary = Some(ManifestEdgeTypeDictionaryRef {
            path: "relationship-types-00000000000000000001.bin".to_string(),
            checksum: "crc32:missing".to_string(),
            entry_count: 1,
            bytes: 32,
        });
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        fs::remove_file(dictionary_path).expect("referenced dictionary removes");
        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");
        assert_eq!(plan.action, ProjectionRecoveryAction::FullRebuild);
        assert!(plan
            .reason
            .as_deref()
            .is_some_and(|reason| reason.contains("relationship type")));
    }

    #[test]
    fn recovery_rejects_segment_type_ids_outside_cumulative_dictionary() {
        let dir = ProjectionArtifactDir::new("recovery_rejects_segment_type_outside_dictionary");
        write_file(dir.path().join("base.pggraph"), b"base");
        let dictionary_path = dir
            .path()
            .join("relationship-types-00000000000000000001.bin");
        let dictionary =
            EdgeTypeDictionary::try_from_labels(vec![String::new()]).expect("dictionary validates");
        let (checksum, bytes) = write_edge_type_dictionary_artifact(
            dir.path(),
            &dictionary_path,
            &dictionary,
            &test_governor(),
        )
        .expect("dictionary writes");
        let segment_path = dir.path().join("active.pggraph-delta");
        edge_segment(1, 0, &[(0, 1, 1)])
            .write_to_path(&segment_path)
            .expect("segment writes");
        let mut manifest = base_manifest(1);
        manifest.edge_type_dictionary = Some(ManifestEdgeTypeDictionaryRef {
            path: relative_path(dir.path(), &dictionary_path),
            checksum,
            entry_count: 1,
            bytes,
        });
        manifest.segments.push(segment_ref(
            dir.path(),
            &segment_path,
            &checksum_for_path(&segment_path),
        ));
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        let constrained =
            crate::resource::ResourceGovernor::new(crate::resource::ResourceLimits::memory_only(
                crate::resource::MemoryBudget::new(crate::resource::ByteCount::from_bytes(1)),
            ));
        assert!(matches!(
            validate_manifest_edge_type_dictionary_governed(
                dir.path(),
                &manifest,
                &[],
                &[],
                None,
                &constrained,
            ),
            Err(GraphError::ResourceLimit { .. })
        ));
        validate_manifest_edge_type_dictionary_governed(
            dir.path(),
            &manifest,
            &[],
            &[],
            None,
            &test_governor(),
        )
        .expect("sufficient recovery budget validates dictionary");
        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");
        assert_eq!(plan.action, ProjectionRecoveryAction::FullRebuild);
        assert!(plan
            .reason
            .as_deref()
            .is_some_and(|reason| reason.contains("outside the cumulative dictionary")));

        let chunk_path = dir.path().join("active.pggraph-chunk");
        edge_segment(2, 0, &[(0, 1, 1)])
            .write_to_path(&chunk_path)
            .expect("chunk writes");
        let mut chunk_manifest = base_manifest(2);
        chunk_manifest.edge_type_dictionary = manifest.edge_type_dictionary.clone();
        chunk_manifest.base_chunks.push(ManifestChunkRef {
            path: relative_path(dir.path(), &chunk_path),
            checksum: checksum_for_path(&chunk_path),
            source_start: 0,
            source_end: 1,
            dirty_source_count: 1,
            dirty_edge_count: 1,
        });
        ProjectionManifestStore::new(dir.path())
            .publish_if_current(&chunk_manifest, Some(1))
            .expect("chunk manifest publishes");
        let chunk_plan = plan_projection_recovery(dir.path()).expect("chunk recovery plans");
        assert_eq!(chunk_plan.action, ProjectionRecoveryAction::FullRebuild);
        assert!(chunk_plan
            .reason
            .as_deref()
            .is_some_and(|reason| reason.contains("outside the cumulative dictionary")));
    }

    #[test]
    fn artifact_recovery_rejects_dictionary_prefix_and_base_registry_overflow() {
        use crate::engine::Engine;
        use crate::persistence::write_graph_file;

        let dir = ProjectionArtifactDir::new(
            "artifact_recovery_rejects_dictionary_prefix_and_base_registry_overflow",
        );
        let graph_path = dir.path().join("base.pggraph");
        let mut engine = Engine::new();
        engine
            .register_edge_type("base")
            .expect("base label registers");
        engine.finish_build(None);
        write_graph_file(&engine, &graph_path).expect("base graph writes");

        let dictionary_path = dir
            .path()
            .join("relationship-types-00000000000000000001.bin");
        let dictionary =
            EdgeTypeDictionary::try_from_labels(vec![String::new(), "different".to_string()])
                .expect("dictionary validates independently");
        let (checksum, bytes) = write_edge_type_dictionary_artifact(
            dir.path(),
            &dictionary_path,
            &dictionary,
            &test_governor(),
        )
        .expect("dictionary writes");
        let mut manifest = ProjectionManifest::base_only(
            1,
            "base.pggraph",
            graph_artifact_checksum_for_path(&graph_path).expect("base checksum reads"),
            graph_artifact_version(),
            0,
            1,
        );
        manifest.edge_type_dictionary = Some(ManifestEdgeTypeDictionaryRef {
            path: relative_path(dir.path(), &dictionary_path),
            checksum,
            entry_count: 2,
            bytes,
        });
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        let prefix_plan = plan_projection_recovery_for_artifact(dir.path(), Some(&graph_path))
            .expect("prefix recovery plans");
        assert_eq!(prefix_plan.action, ProjectionRecoveryAction::FullRebuild);
        assert!(prefix_plan
            .reason
            .as_deref()
            .is_some_and(|reason| reason.contains("base registry prefix")));

        let segment_path = dir.path().join("outside-base.pggraph-delta");
        edge_segment(2, 0, &[(0, 1, 2)])
            .write_to_path(&segment_path)
            .expect("segment writes");
        let mut no_dictionary = ProjectionManifest::base_only(
            2,
            "base.pggraph",
            graph_artifact_checksum_for_path(&graph_path).expect("base checksum reads"),
            graph_artifact_version(),
            0,
            2,
        );
        no_dictionary.segments.push(segment_ref(
            dir.path(),
            &segment_path,
            &checksum_for_path(&segment_path),
        ));
        ProjectionManifestStore::new(dir.path())
            .publish_if_current(&no_dictionary, Some(1))
            .expect("second manifest publishes");
        let range_plan = plan_projection_recovery_for_artifact(dir.path(), Some(&graph_path))
            .expect("range recovery plans");
        assert_eq!(range_plan.action, ProjectionRecoveryAction::FullRebuild);
        assert!(range_plan
            .reason
            .as_deref()
            .is_some_and(|reason| reason.contains("cumulative dictionary")));
    }

    #[test]
    fn generation_gc_retains_referenced_edge_type_dictionaries() {
        let dir = ProjectionArtifactDir::new("generation_gc_retains_edge_type_dictionary");
        write_file(dir.path().join("base.pggraph"), b"base");
        let dictionary_path = dir
            .path()
            .join("relationship-types-00000000000000000001.bin");
        write_file(&dictionary_path, b"dictionary");
        let mut manifest = base_manifest(1);
        manifest.edge_type_dictionary = Some(ManifestEdgeTypeDictionaryRef {
            path: relative_path(dir.path(), &dictionary_path),
            checksum: "crc32:dictionary".to_string(),
            entry_count: 1,
            bytes: 10,
        });
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        let mut protected = BTreeSet::new();
        insert_active_manifest_references(dir.path(), &manifest, &mut protected)
            .expect("references resolve");
        assert!(protected.contains(&dictionary_path));
    }

    #[test]
    fn failed_dictionary_candidate_preserves_the_current_generation() {
        let dir = ProjectionArtifactDir::new("failed_dictionary_candidate_preserves_current");
        write_file(dir.path().join("base.pggraph"), b"base");
        let store = ProjectionManifestStore::new(dir.path());
        store.publish(&base_manifest(1)).expect("current publishes");
        let candidate = dir
            .path()
            .join("relationship-types-00000000000000000002.bin");
        write_file(&candidate, b"corrupt candidate");
        let removed = cleanup_interrupted_replacement(dir.path(), Some(1), Some(2))
            .expect("candidate cleanup succeeds");
        assert_eq!(removed, 1);
        assert!(!candidate.exists());
        assert_eq!(
            store.current_generation_id().expect("current reads"),
            Some(1)
        );
    }

    #[test]
    fn load_missing_referenced_segment_is_rejected() {
        let dir = ProjectionArtifactDir::new("load_missing_referenced_segment_is_rejected");
        write_file(dir.path().join("base.pggraph"), b"base");
        let mut manifest = base_manifest(1);
        manifest.segments.push(ManifestSegmentRef {
            path: "missing.pggraph-delta".to_string(),
            checksum: "crc32:missing".to_string(),
            level: 0,
            source_start: 0,
            source_end: 1,
            sync_watermark: 1,
        });
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect_err("missing referenced segment rejects");
    }

    #[test]
    fn load_missing_unref_temp_segment_is_ignored() {
        let dir = ProjectionArtifactDir::new("load_missing_unref_temp_segment_is_ignored");
        write_file(dir.path().join("base.pggraph"), b"base");
        write_file(
            dir.path()
                .join("projection-generation-00000000000000000003-segment-00000000.tmp"),
            b"partial",
        );
        ProjectionManifestStore::new(dir.path())
            .publish(&base_manifest(1))
            .expect("manifest publishes");

        let loaded = validate_active_projection(dir.path())
            .expect("validation ignores temp")
            .expect("manifest exists");

        assert_eq!(loaded.generation_id, 1);
    }

    #[test]
    fn base_chunk_corruption_repairs_from_postgresql() {
        let dir = ProjectionArtifactDir::new("base_chunk_corruption_repairs_from_postgresql");
        write_file(dir.path().join("base.pggraph"), b"base");
        let source = edge_store_from_tuples(3, &[(0, 1, 1), (1, 2, 1)]);
        let chunk_path = dir.path().join("active.pggraph-chunk");
        let chunk = edge_segment(1, 0, &[(0, 1, 1)]);
        chunk.write_to_path(&chunk_path).expect("chunk writes");
        let checksum = checksum_for_path(&chunk_path);
        let mut manifest = base_manifest(1);
        manifest.base_chunks.push(ManifestChunkRef {
            path: relative_path(dir.path(), &chunk_path),
            checksum,
            source_start: 0,
            source_end: 2,
            dirty_source_count: 2,
            dirty_edge_count: 2,
        });
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        write_file(&chunk_path, b"corrupt");

        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");
        assert_eq!(plan.action, ProjectionRecoveryAction::TargetedChunkRepair);
        assert_eq!(plan.generation_id, Some(1));

        let repaired = repair_active_base_chunks(dir.path(), &EdgeStoreChunkSource::new(&source))
            .expect("chunk repair runs")
            .expect("chunk repair publishes")
            .manifest;

        assert_eq!(repaired.previous_generation_id, Some(1));
        assert_eq!(repaired.base_chunks.len(), 1);
        assert_ne!(repaired.base_chunks[0].path, manifest.base_chunks[0].path);
    }

    #[test]
    fn missing_base_chunk_repairs_from_postgresql_when_metadata_validates() {
        let dir = ProjectionArtifactDir::new(
            "missing_base_chunk_repairs_from_postgresql_when_metadata_validates",
        );
        write_file(dir.path().join("base.pggraph"), b"base");
        let source = edge_store_from_tuples(3, &[(0, 1, 1), (1, 2, 1)]);
        let chunk_path = dir.path().join("missing.pggraph-chunk");
        let mut manifest = base_manifest(1);
        manifest.base_chunks.push(ManifestChunkRef {
            path: relative_path(dir.path(), &chunk_path),
            checksum: "crc32:missing".to_string(),
            source_start: 0,
            source_end: 2,
            dirty_source_count: 2,
            dirty_edge_count: 2,
        });
        write_file(&chunk_path, b"placeholder");
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");
        fs::remove_file(&chunk_path).expect("chunk file removed");

        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");
        assert_eq!(plan.action, ProjectionRecoveryAction::TargetedChunkRepair);
        assert_eq!(plan.generation_id, Some(1));

        let repaired = repair_active_base_chunks(dir.path(), &EdgeStoreChunkSource::new(&source))
            .expect("chunk repair runs")
            .expect("chunk repair publishes")
            .manifest;

        assert_eq!(repaired.previous_generation_id, Some(1));
        assert_eq!(repaired.base_chunks.len(), 1);
        assert!(dir.path().join(&repaired.base_chunks[0].path).exists());
    }

    #[test]
    fn corrupt_manifest_triggers_full_projection_rebuild() {
        let dir = ProjectionArtifactDir::new("corrupt_manifest_triggers_full_projection_rebuild");
        write_file(dir.path().join(manifest_file_name(3)), b"{not json");

        let plan = plan_projection_recovery(dir.path()).expect("recovery plans");

        assert_eq!(plan.action, ProjectionRecoveryAction::FullRebuild);
        assert_eq!(plan.generation_id, Some(3));
    }

    #[test]
    fn full_rebuild_restores_valid_projection_generation() {
        use crate::engine::Engine;
        use crate::persistence::write_graph_file;

        let dir = ProjectionArtifactDir::new("full_rebuild_restores_valid_projection_generation");
        let graph_path = dir.path().join("main.pggraph");
        let mut engine = Engine::new();
        engine.finish_build(None);
        write_graph_file(&engine, &graph_path).expect("base graph writes");
        write_file(dir.path().join(manifest_file_name(4)), b"{not json");

        let generation_id = next_rebuild_generation_id(dir.path()).expect("next generation id");
        quarantine_latest_manifest(dir.path()).expect("corrupt manifest quarantines");
        let manifest = publish_rebuilt_base_manifest(&graph_path, 42)
            .expect("rebuilt projection manifest publishes");
        let loaded = validate_active_projection(dir.path())
            .expect("rebuilt generation validates")
            .expect("rebuilt manifest exists");

        assert_eq!(generation_id, 5);
        assert_eq!(manifest.generation_id, 5);
        assert_eq!(loaded.generation_id, 5);
        assert_eq!(loaded.base_artifact_path, "main.pggraph");
        assert_eq!(loaded.sync_watermark, 42);
    }

    #[test]
    fn v7_recovery_uses_actual_version_and_width() {
        use crate::edge_store::EdgeTypeWidth;
        use crate::engine::Engine;
        use crate::persistence::{
            graph_artifact_metadata_for_path, write_graph_file, write_v6_graph_file_for_test,
        };

        let dir = ProjectionArtifactDir::new("recovery_validates_actual_base_artifact_version");
        let mut engine = Engine::new();
        engine.finish_build(None);

        let v6_path = dir.path().join("base-v6.pggraph");
        write_v6_graph_file_for_test(&engine, &v6_path).expect("v6 base writes");
        let v6_metadata = graph_artifact_metadata_for_path(&v6_path).expect("v6 metadata reads");
        assert_eq!(v6_metadata.version, 6);
        assert_eq!(v6_metadata.edge_type_width, EdgeTypeWidth::One);
        crate::persistence::load_graph_file(&v6_path).expect("v6 base reloads");
        let v6_manifest =
            publish_rebuilt_base_manifest(&v6_path, 41).expect("recovery publishes v6 manifest");
        assert_eq!(v6_manifest.base_artifact_version, 6);
        validate_manifest_base_metadata(&v6_path, &v6_manifest)
            .expect("recovery validates parsed v6 metadata");

        let mut wide_engine = Engine::new();
        for type_id in 1..=255 {
            wide_engine
                .register_edge_type(&format!("type_{type_id}"))
                .expect("wide registry grows");
        }
        wide_engine.finish_build(None);
        let v7_path = dir.path().join("base-v7.pggraph");
        write_graph_file(&wide_engine, &v7_path).expect("v7 base writes");
        let v7_metadata = graph_artifact_metadata_for_path(&v7_path).expect("v7 metadata reads");
        assert_eq!(v7_metadata.version, 7);
        assert_eq!(v7_metadata.edge_type_width, EdgeTypeWidth::Two);
        crate::persistence::load_graph_file(&v7_path).expect("v7 base reloads");
        let v7_manifest =
            publish_rebuilt_base_manifest(&v7_path, 42).expect("recovery publishes v7 manifest");
        assert_eq!(v7_manifest.base_artifact_version, 7);
        validate_manifest_base_metadata(&v7_path, &v7_manifest)
            .expect("recovery validates parsed v7 metadata");
    }

    #[test]
    fn healthy_rebuild_plan_uses_a_generation_specific_private_base_path() {
        let dir = ProjectionArtifactDir::new(
            "healthy_rebuild_plan_uses_a_generation_specific_private_base_path",
        );
        let plan = plan_generation_specific_rebuilt_base(dir.path())
            .expect("healthy rebuilt-base generation plans");

        assert_eq!(plan.expected_current_generation(), None);
        assert_eq!(plan.generation_id(), 1);
        assert_eq!(
            plan.candidate_base_name(),
            "projection-generation-00000000000000000001-base.pggraph"
        );
        assert_eq!(
            plan.candidate_base_path(),
            dir.path().join(plan.candidate_base_name())
        );
    }

    #[test]
    fn orphan_rebuilt_base_reserves_generation_after_pre_manifest_crash() {
        let dir = ProjectionArtifactDir::new(
            "orphan_rebuilt_base_reserves_generation_after_pre_manifest_crash",
        );
        let interrupted = plan_generation_specific_rebuilt_base(dir.path())
            .expect("interrupted generation plans");
        write_file(interrupted.candidate_base_path(), b"fsynced candidate");

        let resumed = plan_generation_specific_rebuilt_base(dir.path())
            .expect("post-crash generation replans");

        assert_eq!(interrupted.generation_id(), 1);
        assert_eq!(resumed.generation_id(), 2);
        assert_ne!(
            resumed.candidate_base_path(),
            interrupted.candidate_base_path()
        );
    }

    #[test]
    fn rebuilt_base_sidecars_also_reserve_generation_ids() {
        let dir = ProjectionArtifactDir::new("rebuilt_base_sidecars_reserve_generation_ids");
        write_file(
            dir.path()
                .join("projection-generation-00000000000000000008-base.pggraph.projection_mode"),
            b"csr_readonly",
        );

        assert_eq!(
            next_rebuild_generation_id(dir.path()).expect("sidecar generation scans"),
            9
        );
    }

    #[test]
    fn healthy_rebuild_manifest_retains_previous_base_as_obsolete() {
        use crate::engine::Engine;
        use crate::persistence::write_graph_file;

        let dir = ProjectionArtifactDir::new(
            "healthy_rebuild_manifest_retains_previous_base_as_obsolete",
        );
        let old_path = dir
            .path()
            .join("projection-generation-00000000000000000001-base.pggraph");
        let mut engine = Engine::new();
        engine.finish_build(None);
        write_graph_file(&engine, &old_path).expect("old base writes");
        let mut previous = ProjectionManifest::base_only(
            1,
            old_path.file_name().unwrap().to_string_lossy(),
            graph_artifact_checksum_for_path(&old_path).expect("old checksum reads"),
            graph_artifact_version(),
            7,
            1,
        );
        previous.last_ingestion_unix_micros = Some(123);
        ProjectionManifestStore::new(dir.path())
            .publish(&previous)
            .expect("previous generation publishes");

        let plan = plan_generation_specific_rebuilt_base(dir.path())
            .expect("replacement generation plans");
        write_graph_file(&engine, plan.candidate_base_path()).expect("candidate base writes");
        let manifest = prepare_generation_specific_rebuilt_base_manifest(&plan, 9)
            .expect("replacement manifest prepares");
        assert_eq!(
            ProjectionManifestStore::new(dir.path())
                .current_generation_id()
                .expect("current generation reads"),
            Some(1)
        );
        publish_prepared_generation_specific_rebuilt_base(&plan, &manifest)
            .expect("replacement generation publishes");

        assert_eq!(manifest.manifest().previous_generation_id, Some(1));
        assert_eq!(manifest.manifest().last_ingestion_unix_micros, Some(123));
        assert!(manifest.manifest().obsolete_files.iter().any(|reference| {
            reference.path == previous.base_artifact_path
                && reference.bytes == fs::metadata(&old_path).unwrap().len()
        }));
    }

    #[test]
    fn prepared_rebuild_publish_rejects_candidate_changed_after_preparation() {
        use crate::engine::Engine;
        use crate::persistence::write_graph_file;

        let dir = ProjectionArtifactDir::new(
            "prepared_rebuild_publish_rejects_plan_and_checksum_mismatches",
        );
        let base_path = dir.path().join("base.pggraph");
        let mut engine = Engine::new();
        engine.finish_build(None);
        write_graph_file(&engine, &base_path).expect("base writes");
        let first = ProjectionManifest::base_only(
            1,
            "base.pggraph",
            graph_artifact_checksum_for_path(&base_path).expect("base checksum reads"),
            graph_artifact_version(),
            1,
            1,
        );
        let store = ProjectionManifestStore::new(dir.path());
        store.publish(&first).expect("first generation publishes");
        let plan = plan_generation_specific_rebuilt_base(dir.path())
            .expect("replacement generation plans");
        write_graph_file(&engine, plan.candidate_base_path()).expect("candidate base writes");
        let prepared =
            prepare_generation_specific_rebuilt_base_manifest(&plan, 2).expect("manifest prepares");

        use std::io::Write;
        std::fs::OpenOptions::new()
            .append(true)
            .open(plan.candidate_base_path())
            .expect("candidate opens")
            .write_all(&[0])
            .expect("candidate mutates");
        assert!(matches!(
            publish_prepared_generation_specific_rebuilt_base(&plan, &prepared),
            Err(GraphError::CorruptFile { .. })
        ));
        assert_eq!(
            store
                .current_generation_id()
                .expect("current generation reads"),
            Some(1)
        );
        write_graph_file(&engine, plan.candidate_base_path()).expect("candidate rewrites");
        let prepared = prepare_generation_specific_rebuilt_base_manifest(&plan, 2)
            .expect("replacement manifest prepares again");
        publish_prepared_generation_specific_rebuilt_base(&plan, &prepared)
            .expect("matching manifest publishes");
    }

    #[test]
    fn healthy_rebuild_cas_loser_preserves_the_winning_generation() {
        use crate::engine::Engine;
        use crate::persistence::write_graph_file;

        let dir = ProjectionArtifactDir::new(
            "healthy_rebuild_cas_loser_preserves_the_winning_generation",
        );
        let base_path = dir.path().join("base.pggraph");
        let mut engine = Engine::new();
        engine.finish_build(None);
        write_graph_file(&engine, &base_path).expect("base writes");
        let first = ProjectionManifest::base_only(
            1,
            "base.pggraph",
            graph_artifact_checksum_for_path(&base_path).expect("base checksum reads"),
            graph_artifact_version(),
            1,
            1,
        );
        let store = ProjectionManifestStore::new(dir.path());
        store.publish(&first).expect("first generation publishes");
        let loser =
            plan_generation_specific_rebuilt_base(dir.path()).expect("losing generation plans");
        write_graph_file(&engine, loser.candidate_base_path()).expect("loser base writes");

        let winner = ProjectionManifest::base_only(
            3,
            "base.pggraph",
            graph_artifact_checksum_for_path(&base_path).expect("winner checksum reads"),
            graph_artifact_version(),
            2,
            2,
        );
        store
            .publish_if_current(&winner, Some(1))
            .expect("winner publishes");

        assert!(matches!(
            publish_generation_specific_rebuilt_base(&loser, 3),
            Err(GraphError::BuildLocked)
        ));
        assert_eq!(
            store
                .load_latest_current()
                .expect("current reads")
                .expect("winner remains")
                .generation_id,
            3
        );
    }

    #[test]
    fn repeated_rebuild_rebases_checksum_and_obsoletes_projection_files() {
        use crate::engine::Engine;
        use crate::persistence::{load_graph_file, write_graph_file};

        let dir = ProjectionArtifactDir::new(
            "repeated_rebuild_rebases_checksum_and_obsoletes_projection_files",
        );
        let graph_path = dir.path().join("main.pggraph");
        let mut engine = Engine::new();
        engine.finish_build(None);
        write_graph_file(&engine, &graph_path).expect("initial base graph writes");

        let segment_path = dir.path().join("superseded.pggraph-delta");
        edge_segment(1, 0, &[(0, 0, 1)])
            .write_to_path(&segment_path)
            .expect("superseded segment writes");
        let mut previous = ProjectionManifest::base_only(
            1,
            "main.pggraph",
            graph_artifact_checksum_for_path(&graph_path).expect("initial checksum reads"),
            graph_artifact_version(),
            7,
            1,
        );
        previous.last_ingestion_unix_micros = Some(123);
        previous
            .segments
            .push(segment_ref(dir.path(), &segment_path, "crc32:00000000"));
        ProjectionManifestStore::new(dir.path())
            .publish(&previous)
            .expect("previous manifest publishes");
        assert_eq!(
            plan_projection_recovery_for_artifact(dir.path(), Some(&graph_path))
                .expect("corrupt segment recovery plans")
                .action,
            ProjectionRecoveryAction::FullRebuild
        );

        crate::sync::sync_insert(&mut engine, 42, "post-build", None)
            .expect("replacement base changes");
        engine.edge_store = crate::edge_store::EdgeStore::from_edges(
            engine.node_store.node_count(),
            Vec::new(),
            false,
        );
        engine.record_applied_sync_id(9);
        write_graph_file(&engine, &graph_path).expect("replacement base graph writes");
        let rebased =
            publish_rebuilt_base_manifest(&graph_path, 9).expect("replacement manifest publishes");
        let loaded = load_graph_file(&graph_path).expect("rebased base reloads");

        assert_eq!(rebased.generation_id, 2);
        assert_eq!(rebased.previous_generation_id, Some(1));
        assert_eq!(rebased.sync_watermark, 9);
        assert_eq!(rebased.last_ingestion_unix_micros, Some(123));
        assert!(rebased.segments.is_empty());
        assert!(rebased.base_chunks.is_empty());
        assert!(rebased.relationship_identities.is_none());
        assert_eq!(rebased.obsolete_files.len(), 1);
        assert_eq!(rebased.obsolete_files[0].path, "superseded.pggraph-delta");
        assert_eq!(
            rebased.base_artifact_checksum,
            graph_artifact_checksum_for_path(&graph_path).expect("replacement checksum reads")
        );
        assert_eq!(loaded.applied_sync_id, 9);
        assert!(loaded.resolve(42, "post-build").is_some());
    }

    #[test]
    fn stale_base_artifact_checksum_triggers_full_projection_rebuild() {
        use crate::engine::Engine;
        use crate::persistence::write_graph_file;

        let dir = ProjectionArtifactDir::new(
            "stale_base_artifact_checksum_triggers_full_projection_rebuild",
        );
        let graph_path = dir.path().join("main.pggraph");
        let mut engine = Engine::new();
        engine.finish_build(None);
        write_graph_file(&engine, &graph_path).expect("base graph writes");
        let manifest = ProjectionManifest::base_only(1, "main.pggraph", "crc32:stale", 1, 1, 1);
        ProjectionManifestStore::new(dir.path())
            .publish(&manifest)
            .expect("manifest publishes");

        let plan = plan_projection_recovery_for_artifact(dir.path(), Some(&graph_path))
            .expect("recovery plans");

        assert_eq!(plan.action, ProjectionRecoveryAction::FullRebuild);
        assert_eq!(plan.generation_id, Some(1));
    }

    fn base_manifest(generation_id: u64) -> ProjectionManifest {
        ProjectionManifest::base_only(generation_id, "base.pggraph", "crc32:base", 1, 1, 1)
    }

    fn edge_segment(
        generation_id: u64,
        source_start: u32,
        edges: &[(u32, u32, u8)],
    ) -> DeltaSegment {
        let source_end = edges
            .iter()
            .map(|(source, _, _)| source + 1)
            .max()
            .unwrap_or(source_start + 1);
        let mut segment = DeltaSegment::new(
            SegmentKind::Edge,
            0,
            TraversalDirection::Out,
            source_start,
            source_end,
            i64::try_from(generation_id).expect("generation fits i64"),
        )
        .expect("segment creates");
        for &(source, target, type_id) in edges {
            segment.edge_inserts.push(SegmentEdge {
                source,
                target,
                type_id: crate::types::EdgeTypeId::test_v6(type_id),
                schema_reversed: false,
                relationship_id: None,
            });
        }
        segment
    }

    fn segment_ref(root: &Path, path: &Path, checksum: &str) -> ManifestSegmentRef {
        ManifestSegmentRef {
            path: relative_path(root, path),
            checksum: checksum.to_string(),
            level: 0,
            source_start: 0,
            source_end: 1,
            sync_watermark: 1,
        }
    }

    fn relative_path(root: &Path, path: &Path) -> String {
        path.strip_prefix(root)
            .expect("path is under root")
            .to_string_lossy()
            .into_owned()
    }

    fn checksum_for_path(path: &Path) -> String {
        format!(
            "crc32:{:08x}",
            crc32fast::hash(&fs::read(path).expect("file reads"))
        )
    }

    fn write_file(path: impl Into<PathBuf>, bytes: &[u8]) {
        fs::write(path.into(), bytes).expect("file writes");
    }
}
