use super::admin::{check_enabled_result, require_graph_admin_result, with_panic_boundary};
use super::*;

#[cfg(feature = "development")]
thread_local! {
    static QUERY_START_PROBE_COUNTS: std::cell::Cell<(u32, u32)> =
        const { std::cell::Cell::new((0, 0)) };
}

#[cfg(feature = "development")]
pub(crate) fn reset_query_start_probe_counts() {
    QUERY_START_PROBE_COUNTS.set((0, 0));
}

#[cfg(feature = "development")]
pub(crate) fn query_start_probe_counts() -> (u32, u32) {
    QUERY_START_PROBE_COUNTS.get()
}

#[cfg(feature = "development")]
pub(crate) fn record_query_start_catalog_read() {
    QUERY_START_PROBE_COUNTS.with(|counts| {
        let (catalog_reads, pending_probes) = counts.get();
        counts.set((catalog_reads.saturating_add(1), pending_probes));
    });
}

#[cfg(feature = "development")]
pub(crate) fn record_query_start_pending_probe() {
    QUERY_START_PROBE_COUNTS.with(|counts| {
        let (catalog_reads, pending_probes) = counts.get();
        counts.set((catalog_reads, pending_probes.saturating_add(1)));
    });
}

/// Reset the engine — clear graph and remove persisted files.
#[pg_extern(schema = "graph", security_definer)]
#[search_path(pg_catalog, pg_temp)]
fn reset() {
    reset_selected_graph(false);
}

/// Reset the engine, optionally clearing selected-graph registrations.
#[pg_extern(schema = "graph", name = "reset", security_definer)]
#[search_path(pg_catalog, pg_temp)]
fn reset_with_registrations(clear_registrations: bool) {
    reset_selected_graph(clear_registrations);
}

fn reset_selected_graph(clear_registrations: bool) {
    with_panic_boundary("reset()", move || {
        require_graph_admin_result().unwrap_or_else(|err| err.report());
        crate::projection::tx_delta::ensure_engine_replacement_allowed("graph.reset()")
            .unwrap_or_else(|err| err.report());

        let caller_oid = catalog::current_role_oid().unwrap_or_else(|err| err.report());
        let graph = catalog::selected_or_default_graph_metadata_for_role(caller_oid)
            .unwrap_or_else(|err| err.report());
        let cleared: Option<catalog::ClearedRegistrationCounts> = clear_registrations.then(|| {
            catalog::clear_graph_registrations(&graph.graph_id).unwrap_or_else(|err| err.report())
        });

        ENGINE.with(|e| {
            *e.borrow_mut() = Engine::new();
        });
        crate::runtime_state::clear_loaded_graph();
        crate::runtime_state::clear_replacement_recovery_for(&graph.graph_id);
        persistence::remove_graph_artifacts_for(&graph.graph_id).unwrap_or_else(|err| err.report());
        if let Some(cleared) = cleared {
            pgrx::notice!(
                "graph: removed persisted files and cleared {} tables, {} edges, and {} filter columns for graph {} ({})",
                cleared.tables,
                cleared.edges,
                cleared.filter_columns,
                graph.graph_name,
                graph.graph_id
            );
        } else {
            pgrx::notice!(
                "graph: removed persisted files for graph {} ({})",
                graph.graph_name,
                graph.graph_id
            );
        }
    })
}

#[pg_extern(schema = "graph", security_definer)]
#[search_path(pg_catalog, pg_temp)]
fn select_graph(
    graph_name: &str,
    tenant: default!(Option<&str>, "NULL"),
    namespace: default!(Option<&str>, "NULL"),
) -> TableIterator<
    'static,
    (
        name!(graph_id, String),
        name!(graph_name, String),
        name!(loaded, bool),
    ),
> {
    with_panic_boundary("select_graph()", || {
        let caller_oid = catalog::current_role_oid().unwrap_or_else(|err| err.report());
        let graph =
            resolve_visible_runtime_graph_for_role(graph_name, tenant, namespace, caller_oid);
        crate::projection::tx_delta::ensure_engine_replacement_allowed("graph.select_graph()")
            .unwrap_or_else(|err| err.report());
        catalog::set_selected_graph_id(&graph.graph_id).unwrap_or_else(|err| err.report());
        let mut loaded = crate::runtime_state::selected_graph_matches_loaded_slot(&graph.graph_id);
        if !loaded && crate::runtime_state::loaded_graph_id().is_some() {
            ENGINE.with(|engine| {
                *engine.borrow_mut() = Engine::new();
            });
            crate::runtime_state::clear_loaded_graph();
        }
        if !loaded && graph.residency == "hot" && config::HOT_EAGER_LOAD.get() {
            if let Err(err) =
                load_selected_graph_from_disk(&graph, true, CatalogFingerprintForLoad::Resolve)
            {
                pgrx::warning!("graph: hot eager-load skipped: {}", err);
            }
            loaded = crate::runtime_state::selected_graph_matches_loaded_slot(&graph.graph_id);
        }
        TableIterator::new(vec![(graph.graph_id, graph.graph_name, loaded)])
    })
}

#[pg_extern(schema = "graph", security_definer)]
#[search_path(pg_catalog, pg_temp)]
#[allow(
    clippy::type_complexity,
    reason = "pgrx TableIterator tuple defines this SQL function's ABI"
)]
fn load_graph(
    graph_name: &str,
    tenant: default!(Option<&str>, "NULL"),
    namespace: default!(Option<&str>, "NULL"),
) -> TableIterator<
    'static,
    (
        name!(graph_id, String),
        name!(graph_name, String),
        name!(loaded, bool),
        name!(node_count, Option<i64>),
        name!(edge_count, Option<i64>),
        name!(memory_used_mb, Option<f64>),
        name!(projection_mode, Option<String>),
    ),
> {
    with_panic_boundary("load_graph()", || {
        let caller_oid = catalog::current_role_oid().unwrap_or_else(|err| err.report());
        let graph =
            resolve_visible_runtime_graph_for_role(graph_name, tenant, namespace, caller_oid);
        catalog::require_graph_privilege_for_role(
            &graph,
            catalog::GraphPrivilege::Admin,
            caller_oid,
        )
        .unwrap_or_else(|err| err.report());
        catalog::set_selected_graph_id(&graph.graph_id).unwrap_or_else(|err| err.report());
        load_selected_graph_from_disk(&graph, false, CatalogFingerprintForLoad::Resolve)
            .unwrap_or_else(|err| err.report());
        let snapshot =
            ENGINE.with(|engine| crate::runtime_state::loaded_graph_snapshot(&engine.borrow()));
        let row = snapshot
            .filter(|snapshot| snapshot.graph_id == graph.graph_id)
            .map(|snapshot| {
                (
                    graph.graph_id.clone(),
                    graph.graph_name.clone(),
                    true,
                    Some(snapshot.node_count),
                    Some(snapshot.edge_count),
                    Some(snapshot.memory_used_mb),
                    Some(snapshot.projection_mode),
                )
            })
            .unwrap_or((
                graph.graph_id,
                graph.graph_name,
                false,
                None,
                None,
                None,
                None,
            ));
        TableIterator::new(vec![row])
    })
}

#[pg_extern(schema = "graph", security_definer)]
#[search_path(pg_catalog, pg_temp)]
fn unload_graph(
    graph_name: &str,
    tenant: default!(Option<&str>, "NULL"),
    namespace: default!(Option<&str>, "NULL"),
) -> TableIterator<
    'static,
    (
        name!(graph_id, String),
        name!(graph_name, String),
        name!(unloaded, bool),
    ),
> {
    with_panic_boundary("unload_graph()", || {
        let caller_oid = catalog::current_role_oid().unwrap_or_else(|err| err.report());
        let graph =
            resolve_visible_runtime_graph_for_role(graph_name, tenant, namespace, caller_oid);
        catalog::require_graph_privilege_for_role(
            &graph,
            catalog::GraphPrivilege::Admin,
            caller_oid,
        )
        .unwrap_or_else(|err| err.report());
        let unloaded = crate::runtime_state::selected_graph_matches_loaded_slot(&graph.graph_id);
        if unloaded {
            crate::projection::tx_delta::ensure_engine_replacement_allowed("graph.unload_graph()")
                .unwrap_or_else(|err| err.report());
            ENGINE.with(|engine| {
                *engine.borrow_mut() = Engine::new();
            });
            crate::runtime_state::clear_loaded_graph();
        }
        TableIterator::new(vec![(graph.graph_id, graph.graph_name, unloaded)])
    })
}

#[pg_extern(schema = "graph", security_definer)]
#[search_path(pg_catalog, pg_temp)]
#[allow(
    clippy::type_complexity,
    reason = "pgrx TableIterator tuple defines this SQL function's ABI"
)]
fn loaded_graphs() -> TableIterator<
    'static,
    (
        name!(graph_id, String),
        name!(graph_name, String),
        name!(residency, String),
        name!(node_count, i64),
        name!(edge_count, i64),
        name!(memory_used_mb, f64),
        name!(projection_mode, String),
        name!(last_access_unix_micros, i64),
    ),
> {
    with_panic_boundary("loaded_graphs()", || {
        let caller_oid = catalog::current_role_oid().unwrap_or_else(|err| err.report());
        let visible_graphs = catalog::list_graph_metadata_for_role(caller_oid)
            .unwrap_or_else(|err| err.report())
            .into_iter()
            .filter(|graph| {
                catalog::require_graph_privilege_for_role(
                    graph,
                    catalog::GraphPrivilege::Read,
                    caller_oid,
                )
                .is_ok()
            })
            .collect::<Vec<_>>();
        let row = ENGINE.with(|engine| {
            crate::runtime_state::loaded_graph_snapshot(&engine.borrow()).and_then(
                |mut snapshot| {
                    let graph = visible_graphs
                        .iter()
                        .find(|graph| graph.graph_id == snapshot.graph_id)?;
                    snapshot.residency.clone_from(&graph.residency);
                    Some((
                        snapshot.graph_id,
                        snapshot.graph_name,
                        snapshot.residency,
                        snapshot.node_count,
                        snapshot.edge_count,
                        snapshot.memory_used_mb,
                        snapshot.projection_mode,
                        snapshot.last_access_unix_micros,
                    ))
                },
            )
        });
        TableIterator::new(row.into_iter().collect::<Vec<_>>())
    })
}

#[pg_extern(schema = "graph", security_definer)]
#[search_path(pg_catalog, pg_temp)]
#[allow(
    clippy::type_complexity,
    reason = "pgrx TableIterator tuple defines this SQL function's ABI"
)]
fn graph_runtime_status() -> TableIterator<
    'static,
    (
        name!(graph_id, String),
        name!(graph_name, String),
        name!(residency, String),
        name!(loaded, bool),
        name!(artifact_exists, bool),
        name!(artifact_bytes, Option<i64>),
        name!(node_count, Option<i64>),
        name!(edge_count, Option<i64>),
        name!(memory_used_mb, Option<f64>),
        name!(projection_mode, Option<String>),
        name!(last_access_unix_micros, Option<i64>),
    ),
> {
    with_panic_boundary("graph_runtime_status()", || {
        let caller_oid = catalog::current_role_oid().unwrap_or_else(|err| err.report());
        let loaded =
            ENGINE.with(|engine| crate::runtime_state::loaded_graph_snapshot(&engine.borrow()));
        let rows = catalog::list_graph_metadata_for_role(caller_oid)
            .unwrap_or_else(|err| err.report())
            .into_iter()
            .filter(|graph| {
                catalog::require_graph_privilege_for_role(
                    graph,
                    catalog::GraphPrivilege::Read,
                    caller_oid,
                )
                .is_ok()
            })
            .map(|graph| {
                let logical_artifact =
                    persistence::graph_file_path_for_uncreated(&graph.graph_id).ok();
                let artifact = logical_artifact
                    .as_ref()
                    .and_then(|path| persistence::current_base_artifact_path(path).ok().flatten());
                let artifact_metadata = artifact.as_ref().and_then(|path| path.metadata().ok());
                let artifact_exists = artifact_metadata.is_some();
                let artifact_bytes =
                    artifact_metadata.and_then(|metadata| i64::try_from(metadata.len()).ok());
                let snapshot = loaded
                    .as_ref()
                    .filter(|snapshot| snapshot.graph_id == graph.graph_id);
                (
                    graph.graph_id,
                    graph.graph_name,
                    graph.residency,
                    snapshot.is_some(),
                    artifact_exists,
                    artifact_bytes,
                    snapshot.map(|snapshot| snapshot.node_count),
                    snapshot.map(|snapshot| snapshot.edge_count),
                    snapshot.map(|snapshot| snapshot.memory_used_mb),
                    snapshot.map(|snapshot| snapshot.projection_mode.clone()),
                    snapshot.map(|snapshot| snapshot.last_access_unix_micros),
                )
            })
            .collect::<Vec<_>>();
        TableIterator::new(rows)
    })
}

fn resolve_visible_runtime_graph(
    graph_name: &str,
    tenant: Option<&str>,
    namespace: Option<&str>,
) -> catalog::GraphMetadata {
    catalog::resolve_visible_graph_metadata(graph_name, tenant, namespace)
        .unwrap_or_else(|err| err.report())
        .unwrap_or_else(|| {
            safety::GraphError::InvalidFilter {
                reason: format!("graph '{}' does not exist", graph_name),
            }
            .report()
        })
}

fn resolve_visible_runtime_graph_for_role(
    graph_name: &str,
    tenant: Option<&str>,
    namespace: Option<&str>,
    role_oid: pgrx::pg_sys::Oid,
) -> catalog::GraphMetadata {
    catalog::resolve_visible_graph_metadata_for_role(graph_name, tenant, namespace, role_oid)
        .unwrap_or_else(|err| err.report())
        .unwrap_or_else(|| {
            safety::GraphError::InvalidFilter {
                reason: format!("graph '{}' does not exist", graph_name),
            }
            .report()
        })
}

// ─────────────────────────────────────────────────────────────────────
// Internal helpers
// ─────────────────────────────────────────────────────────────────────

pub(super) fn largest_component_rows(
    limit: i32,
    offset: i32,
    hydrate: bool,
) -> safety::GraphResult<Vec<ComponentNodeRow>> {
    check_enabled_result()?;
    require_graph_admin_result()?;
    let query_start = ensure_current_graph_for_query(current_query_freshness()?)?;
    acl::check_table_acls(
        query_start
            .tables
            .iter()
            .map(|table| table.table_oid)
            .chain(query_start.edges.iter().map(|edge| edge.from_table_oid)),
    )?;
    let offset = usize_from_nonnegative(offset, "offset")?;
    let limit = usize_from_nonnegative(limit, "limit")?;
    let governor = ENGINE.with(|e| e.borrow().analytics_resource_governor())?;
    let coordinator = crate::sql_visibility::prepare_eager_visibility(
        &query_start.tables,
        &query_start.edges,
        &governor,
    )?;
    let context = coordinator.context(&governor);
    let page = ENGINE.with(|e| {
        let eng = e.borrow();
        let cc_result = eng.connected_components_governed_in_context(&context)?;
        let component_id = cc_result
            .component_sizes
            .iter()
            .max_by(|left, right| left.1.cmp(right.1).then_with(|| right.0.cmp(left.0)))
            .map(|(&component_id, _)| component_id as i64)
            .ok_or(safety::GraphError::NotBuilt)?;
        let page_bytes = limit
            .checked_mul(
                std::mem::size_of::<connected_components::ComponentRow>()
                    .saturating_add(eng.node_store.max_primary_key_bytes()),
            )
            .and_then(crate::resource::ByteCount::from_usize)
            .ok_or_else(|| {
                safety::GraphError::Internal("component page estimate overflowed".to_string())
            })?;
        let page_lease = governor
            .reserve_memory(crate::resource::ResourcePhase::QueryBlocking, page_bytes)
            .map_err(crate::safety::resource_limit_error)?;
        let page = connected_components::component_rows_page(
            &cc_result,
            &eng.node_store,
            component_id as u32,
            offset,
            limit,
        )?;
        page_lease.retain_until_governor_drop();
        Ok::<_, safety::GraphError>(page)
    })?;
    hydrate_component_page_governed(page, hydrate, &governor, &query_start.tables)
}

pub(super) fn component_rows(
    component_id: i64,
    limit: i32,
    offset: i32,
    hydrate: bool,
) -> safety::GraphResult<Vec<ComponentNodeRow>> {
    if component_id < 0 {
        return Err(safety::GraphError::InvalidFilter {
            reason: "component_id must be non-negative".to_string(),
        });
    }
    check_enabled_result()?;
    require_graph_admin_result()?;
    let query_start = ensure_current_graph_for_query(current_query_freshness()?)?;
    acl::check_table_acls(
        query_start
            .tables
            .iter()
            .map(|table| table.table_oid)
            .chain(query_start.edges.iter().map(|edge| edge.from_table_oid)),
    )?;
    let offset = usize_from_nonnegative(offset, "offset")?;
    let limit = usize_from_nonnegative(limit, "limit")?;

    let governor = ENGINE.with(|e| e.borrow().analytics_resource_governor())?;
    let coordinator = crate::sql_visibility::prepare_eager_visibility(
        &query_start.tables,
        &query_start.edges,
        &governor,
    )?;
    let context = coordinator.context(&governor);
    let page = ENGINE.with(|e| {
        let eng = e.borrow();
        let cc_result = eng.connected_components_governed_in_context(&context)?;
        let page_bytes = limit
            .checked_mul(
                std::mem::size_of::<connected_components::ComponentRow>()
                    .saturating_add(eng.node_store.max_primary_key_bytes()),
            )
            .and_then(crate::resource::ByteCount::from_usize)
            .ok_or_else(|| {
                safety::GraphError::Internal("component page estimate overflowed".to_string())
            })?;
        let page_lease = governor
            .reserve_memory(crate::resource::ResourcePhase::QueryBlocking, page_bytes)
            .map_err(crate::safety::resource_limit_error)?;
        let page = connected_components::component_rows_page(
            &cc_result,
            &eng.node_store,
            component_id as u32,
            offset,
            limit,
        )?;
        page_lease.retain_until_governor_drop();
        Ok::<_, safety::GraphError>(page)
    })?;

    hydrate_component_page_governed(page, hydrate, &governor, &query_start.tables)
}

pub(super) fn hydrate_component_page(
    page: Vec<connected_components::ComponentRow>,
    hydrate: bool,
) -> safety::GraphResult<Vec<ComponentNodeRow>> {
    let governor = ENGINE.with(|engine| engine.borrow().analytics_resource_governor())?;
    let (tables, _edges, _filter_columns) = read_catalog()?;
    hydrate_component_page_governed(page, hydrate, &governor, &tables)
}

pub(super) fn hydrate_component_page_governed(
    page: Vec<connected_components::ComponentRow>,
    hydrate: bool,
    governor: &crate::resource::ResourceGovernor,
    tables: &[builder::RegisteredTable],
) -> safety::GraphResult<Vec<ComponentNodeRow>> {
    acl::check_table_acls(page.iter().map(|row| row.node_table.0))?;
    let page_bytes = page.iter().try_fold(0usize, |bytes, row| {
        bytes
            .checked_add(std::mem::size_of::<types::TraversalResult>())
            .and_then(|bytes| bytes.checked_add(row.node_id.len()))
            .and_then(|bytes| bytes.checked_add(std::mem::size_of::<ComponentNodeRow>()))
            .ok_or_else(|| {
                safety::GraphError::Internal("component output estimate overflowed".to_string())
            })
    })?;
    let output_lease = governor
        .reserve_memory(
            crate::resource::ResourcePhase::QueryCandidates,
            crate::resource::ByteCount::from_usize(page_bytes).ok_or_else(|| {
                safety::GraphError::Internal(
                    "component output estimate does not fit u64".to_string(),
                )
            })?,
        )
        .map_err(crate::safety::resource_limit_error)?;
    governor
        .consume_work(
            crate::resource::ResourcePhase::QueryCandidates,
            crate::resource::WorkUnits::new(u64::try_from(page.len()).unwrap_or(u64::MAX)),
        )
        .map_err(crate::safety::resource_limit_error)?;
    let traversal_rows = page
        .iter()
        .map(|row| types::TraversalResult {
            node_table: row.node_table,
            node_id: row.node_id.clone(),
            depth: 0,
            path: Vec::new(),
            edge_path: Vec::new(),
        })
        .collect::<Vec<_>>();
    let mut hydrated = if hydrate {
        crate::sql_hydration::hydrate_nodes_governed_with_tables(&traversal_rows, governor, tables)?
    } else {
        HashMap::new()
    };

    let rows = page
        .into_iter()
        .map(|row| {
            let node = hydrated.remove(&(row.node_table.0, row.node_id.clone()));
            (
                row.component_id as i64,
                pgrx::pg_sys::Oid::from_u32(row.node_table.0),
                row.node_id,
                node,
            )
        })
        .collect();
    output_lease.retain_until_governor_drop();
    Ok(rows)
}

/// Auto-load the persisted graph if the engine is empty and auto_load is enabled.
///
/// When a .pggraph file exists, this loads the graph via mmap. NodeStore base
/// arrays, the forward EdgeStore CSR, and the ResolutionIndex are mmap-backed.
/// FilterIndex and the edge type registry are bincode-deserialized into
/// backend-local heap, and the reverse EdgeStore CSR is rebuilt into heap for
/// inbound traversal.
pub(super) fn maybe_auto_load(graph: &catalog::GraphMetadata, catalog_fingerprint: u64) {
    if let Err(err) = clear_loaded_graph_if_mismatched(&graph.graph_id) {
        pgrx::warning!("graph: auto-load skipped: {}", err);
        return;
    }

    if !config::AUTO_LOAD.get() {
        return;
    }

    if let Err(err) = load_selected_graph_from_disk(
        graph,
        true,
        CatalogFingerprintForLoad::Known(catalog_fingerprint),
    ) {
        pgrx::warning!("graph: auto-load skipped: {}", err);
    }
}

pub(super) fn clear_loaded_graph_if_mismatched(graph_id: &str) -> safety::GraphResult<()> {
    if crate::projection::tx_delta::has_provisional_edge_types()
        && crate::runtime_state::loaded_graph_id().is_some_and(|loaded| loaded != graph_id)
    {
        crate::projection::tx_delta::ensure_engine_replacement_allowed(
            "selected graph replacement",
        )?;
    }
    if let Some(loaded_graph_id) = crate::runtime_state::loaded_graph_id() {
        if loaded_graph_id != graph_id {
            ENGINE.with(|engine| {
                *engine.borrow_mut() = Engine::new();
            });
            crate::runtime_state::clear_loaded_graph();
        }
    }
    Ok(())
}

#[derive(Clone, Copy)]
enum CatalogFingerprintForLoad {
    Resolve,
    Known(u64),
    Omit,
}

fn load_selected_graph_from_disk(
    graph: &catalog::GraphMetadata,
    quiet_missing: bool,
    catalog_fingerprint: CatalogFingerprintForLoad,
) -> safety::GraphResult<bool> {
    crate::projection::tx_delta::ensure_engine_replacement_allowed("graph.load_graph()")?;
    if quiet_missing && graph.residency == "cold" {
        return Ok(false);
    }
    ENGINE.with(|e| {
        let eng = e.borrow();
        if eng.built {
            if crate::runtime_state::selected_graph_matches_loaded_slot(&graph.graph_id) {
                crate::runtime_state::touch_loaded_graph(&graph.graph_id);
                return Ok(true);
            }
            drop(eng);
            *e.borrow_mut() = Engine::new();
            crate::runtime_state::clear_loaded_graph();
        } else {
            drop(eng);
        }

        // Check if persisted file exists without creating artifact directories
        // during query-time auto-load or operator load inspection.
        let path = persistence::graph_file_path_for_uncreated(&graph.graph_id)?;
        if !persistence::persisted_graph_exists(&path)? {
            if !quiet_missing {
                return Err(safety::GraphError::NotBuilt);
            }
            return Ok(false);
        }

        catalog::enforce_loaded_graph_quota(1)?;
        if config::MAX_LOADED_GRAPHS_PER_BACKEND.get() < 1 {
            return Err(safety::GraphError::InvalidFilter {
                reason: "graph.max_loaded_graphs_per_backend blocks runtime graph loads"
                    .to_string(),
            });
        }

        // Load from .pggraph file via mmap.
        pgrx::log!("graph: loading from {} (mmap)", path.display());
        match persistence::load_graph_file(&path) {
            Ok(mut loaded_engine) => {
                match catalog_fingerprint {
                    CatalogFingerprintForLoad::Known(catalog_fingerprint) => {
                        loaded_engine.set_catalog_fingerprint(catalog_fingerprint);
                    }
                    CatalogFingerprintForLoad::Resolve => {
                        if let Ok((tables, edges, filters)) = read_catalog() {
                            loaded_engine.set_catalog_fingerprint(super::catalog_fingerprint(
                                &tables, &edges, &filters,
                            ));
                        }
                    }
                    CatalogFingerprintForLoad::Omit => {}
                }
                let nc = loaded_engine.node_store.node_count();
                let ec = loaded_engine.edge_store.edge_count();
                *e.borrow_mut() = loaded_engine;
                crate::runtime_state::mark_loaded_graph(graph);
                pgrx::log!(
                    "graph: loaded {} nodes, {} edges (resolution via mmap, zero-copy)",
                    nc,
                    ec
                );
                Ok(true)
            }
            Err(err) if quiet_missing => {
                pgrx::warning!(
                    "graph: load failed: {:?}. Call graph.build() to reconstruct.",
                    err
                );
                Ok(false)
            }
            Err(err) => Err(err),
        }
    })
}

/// Reconcile a replacement interrupted by a PostgreSQL error or cancellation.
///
/// The manifest pointer is the publication boundary. If it did not move, the
/// old resident engine remains valid. If it moved, or a low-memory replacement
/// evicted the resident engine, reload the currently published artifact before
/// serving another graph operation.
pub(crate) fn reconcile_interrupted_replacement(
    graph: &catalog::GraphMetadata,
) -> safety::GraphResult<()> {
    reconcile_interrupted_replacement_with_catalog_fingerprint(
        graph,
        CatalogFingerprintForLoad::Resolve,
    )
}

fn reconcile_interrupted_replacement_with_catalog_fingerprint(
    graph: &catalog::GraphMetadata,
    catalog_fingerprint: CatalogFingerprintForLoad,
) -> safety::GraphResult<()> {
    // Snapshot ownership is independent of graph selection and no interrupted
    // stack can still use it. Release it at the first recovery boundary even
    // when the caller switched graphs or a newer replacement overwrote state.
    crate::runtime_state::clear_replacement_edge_snapshot();
    let Some(recovery) = crate::runtime_state::replacement_recovery_for(&graph.graph_id) else {
        return Ok(());
    };
    let path = persistence::graph_file_path_for_uncreated(&graph.graph_id)?;
    let root = persistence::projection_manifest_root(&path);
    let store = crate::projection::manifest::ProjectionManifestStore::new(&root);
    let mut current_generation = store.current_generation_id_for_recovery()?;
    let mut cleanup_complete = current_generation != recovery.expected_generation;
    if current_generation == recovery.expected_generation {
        // Candidate generation ids are reused until publication advances the
        // manifest pointer. Serialize cancellation cleanup with every writer
        // before deleting exact-generation artifacts, otherwise a stale
        // backend could remove a different backend's live candidate.
        match crate::sql_build::acquire_build_lock() {
            Ok(()) => {
                crate::projection::recovery::cleanup_interrupted_replacement(
                    &root,
                    recovery.expected_generation,
                    recovery.candidate_generation,
                )?;
                // Publication may have won between the first pointer read and
                // lock acquisition. Decide from the serialized observation.
                current_generation = store.current_generation_id_for_recovery()?;
                cleanup_complete = true;
            }
            Err(safety::GraphError::BuildLocked) => {
                // A remains a valid query snapshot while another backend
                // constructs B. Defer candidate cleanup and retain the marker;
                // ordinary reads must not inherit publisher lock contention.
            }
            Err(err) => return Err(err),
        }
    }
    let persisted_available = persistence::persisted_graph_exists(&path)?;
    let resident_matches = ENGINE.with(|engine| {
        engine.borrow().built
            && crate::runtime_state::selected_graph_matches_loaded_slot(&graph.graph_id)
    });

    match crate::runtime_state::replacement_recovery_action(
        recovery.expected_generation,
        current_generation,
        resident_matches,
        persisted_available,
    ) {
        crate::runtime_state::ReplacementRecoveryAction::PreserveResident => {}
        crate::runtime_state::ReplacementRecoveryAction::ReloadPublished => {
            crate::projection::tx_delta::ensure_engine_replacement_allowed(
                "graph replacement recovery",
            )?;
            ENGINE.with(|engine| {
                *engine.borrow_mut() = Engine::new();
            });
            crate::runtime_state::clear_loaded_graph();
            load_selected_graph_from_disk(graph, false, catalog_fingerprint)?;
        }
        crate::runtime_state::ReplacementRecoveryAction::MissingPublished => {
            return Err(safety::GraphError::Internal(format!(
                "graph replacement recovery for '{}' found no resident or published generation; rebuild the graph with persistence enabled",
                graph.graph_name
            )));
        }
    }

    if cleanup_complete {
        crate::runtime_state::clear_replacement_recovery_for(&graph.graph_id);
    }
    Ok(())
}

#[derive(Debug)]
pub(crate) struct QueryStartState {
    pub(crate) graph: catalog::GraphMetadata,
    pub(crate) tables: Vec<builder::RegisteredTable>,
    pub(crate) edges: Vec<builder::RegisteredEdge>,
    pub(crate) filter_columns: Vec<builder::RegisteredFilterColumn>,
    pub(crate) catalog_fingerprint: u64,
    pub(crate) catalog_state: safety::GraphResult<(u64, Option<String>)>,
    pub(crate) applicable_table_oids: Vec<i32>,
    pub(crate) sync_mode: config::SyncMode,
}

fn load_query_start_state(graph: catalog::GraphMetadata) -> safety::GraphResult<QueryStartState> {
    let (tables, edges, filter_columns) = catalog::read_catalog_for_graph(&graph.graph_id)?;
    let catalog_fingerprint = catalog::catalog_fingerprint(&tables, &edges, &filter_columns);
    let catalog_state = catalog::current_catalog_state_from_rows(&tables, &edges, &filter_columns);
    let applicable_table_oids =
        crate::sql_sync::applicable_table_oids_from_catalog(&tables, &edges);
    let sync_mode = current_sync_mode()?;
    Ok(QueryStartState {
        graph,
        tables,
        edges,
        filter_columns,
        catalog_fingerprint,
        catalog_state,
        applicable_table_oids,
        sync_mode,
    })
}

fn prepare_current_graph(
    require_valid_schema: bool,
    allow_auto_load: bool,
) -> safety::GraphResult<QueryStartState> {
    let graph = catalog::selected_or_default_graph_metadata_via_definer()?;
    clear_loaded_graph_if_mismatched(&graph.graph_id)?;
    catalog::require_selected_graph_privilege_via_definer(catalog::GraphPrivilege::Read)?;
    let query_start = match load_query_start_state(graph.clone()) {
        Ok(query_start) => query_start,
        Err(err) => {
            // Recovery still owns cancellation cleanup when catalog validation
            // fails, but successful query startup must never load the same
            // catalog twice merely to fingerprint a recovered artifact.
            reconcile_interrupted_replacement_with_catalog_fingerprint(
                &graph,
                CatalogFingerprintForLoad::Omit,
            )?;
            return Err(err);
        }
    };
    reconcile_interrupted_replacement_with_catalog_fingerprint(
        &query_start.graph,
        CatalogFingerprintForLoad::Known(query_start.catalog_fingerprint),
    )?;
    if allow_auto_load {
        maybe_auto_load(&query_start.graph, query_start.catalog_fingerprint);
    }

    let disabled = disabled_graph_trigger_count()?;
    let applied_sync_id = ENGINE.with(|e| e.borrow().applied_sync_id);
    let pending = crate::sql_sync::pending_sync_rows_for_query_state(
        applied_sync_id,
        &query_start.graph.graph_id,
        &query_start.applicable_table_oids,
    )?;
    ENGINE.with(|e| {
        let mut eng = e.borrow_mut();
        eng.refresh_observed_state(disabled, pending, &query_start.catalog_state);
        if require_valid_schema && matches!(eng.schema_state, engine::SchemaState::Invalid) {
            return Err(safety::GraphError::Internal(
                eng.invalid_reason
                    .clone()
                    .unwrap_or_else(|| "registered graph schema is invalid".to_string()),
            ));
        }
        Ok::<_, safety::GraphError>(())
    })?;

    if matches!(query_start.sync_mode, config::SyncMode::Trigger) && pending > 0 {
        ENGINE.with(|e| {
            let mut eng = e.borrow_mut();
            eng.mark_syncing();
        });
    }
    Ok(query_start)
}

pub(crate) fn ensure_current_graph() -> safety::GraphResult<QueryStartState> {
    prepare_current_graph(true, true)
}

pub(super) fn refresh_current_graph_status() -> safety::GraphResult<config::SyncMode> {
    let graph = catalog::selected_or_default_graph_metadata_via_definer()?;
    clear_loaded_graph_if_mismatched(&graph.graph_id)?;
    catalog::require_selected_graph_privilege_via_definer(catalog::GraphPrivilege::Read)?;
    let disabled = disabled_graph_trigger_count()?;
    let query_start = match load_query_start_state(graph.clone()) {
        Ok(query_start) => query_start,
        Err(err) => {
            reconcile_interrupted_replacement_with_catalog_fingerprint(
                &graph,
                CatalogFingerprintForLoad::Omit,
            )?;
            let sync_mode = current_sync_mode()?;
            ENGINE.with(|engine| {
                engine
                    .borrow_mut()
                    .refresh_observed_state(disabled, 0, &Err(err));
            });
            return Ok(sync_mode);
        }
    };
    reconcile_interrupted_replacement_with_catalog_fingerprint(
        &query_start.graph,
        CatalogFingerprintForLoad::Known(query_start.catalog_fingerprint),
    )?;
    let applied_sync_id = ENGINE.with(|engine| engine.borrow().applied_sync_id);
    let pending = crate::sql_sync::pending_sync_rows_for_query_state(
        applied_sync_id,
        &query_start.graph.graph_id,
        &query_start.applicable_table_oids,
    )?;
    ENGINE.with(|engine| {
        engine
            .borrow_mut()
            .refresh_observed_state(disabled, pending, &query_start.catalog_state);
    });
    Ok(query_start.sync_mode)
}

pub(super) fn current_query_freshness() -> safety::GraphResult<config::QueryFreshness> {
    config::parsed_query_freshness().ok_or_else(|| safety::GraphError::InvalidFilter {
        reason: format!(
            "unsupported graph.query_freshness '{}'; expected 'off', 'apply_pending_sync', or 'error_on_pending'",
            config::query_freshness()
        ),
    })
}

pub(super) fn ensure_current_graph_for_query(
    freshness: config::QueryFreshness,
) -> safety::GraphResult<QueryStartState> {
    let query_start = ensure_current_graph()?;

    if !matches!(query_start.sync_mode, config::SyncMode::Trigger) {
        return Ok(query_start);
    }

    let pending = ENGINE.with(|e| e.borrow().pending_sync_rows);
    if pending <= 0 {
        return Ok(query_start);
    }

    match freshness {
        config::QueryFreshness::Off => Ok(query_start),
        config::QueryFreshness::ErrorOnPending => Err(safety::GraphError::InvalidFilter {
            reason: format!(
                "topology read has {pending} pending sync row(s); call graph.apply_sync() or set graph.query_freshness = 'apply_pending_sync'"
            ),
        }),
        config::QueryFreshness::ApplyPendingSync => {
            // Transaction-local overlays already provide read-your-own-writes.
            // Applying pending sync here would fold uncommitted trigger rows into
            // the backend-local base projection and make rollback leak until reset.
            if crate::projection::tx_delta::stats().dirty {
                return Ok(query_start);
            }

            let high_watermark = crate::sql_sync::max_sync_log_id_for_query_state(
                &query_start.graph.graph_id,
                &query_start.applicable_table_oids,
            )?;
            crate::sql_sync::apply_sync_to_high_watermark_for_query(
                high_watermark,
                &query_start.graph,
                &query_start.tables,
                &query_start.edges,
                &query_start.filter_columns,
                query_start.catalog_fingerprint,
                &query_start.applicable_table_oids,
            )?;
            let pending = ENGINE.with(|e| e.borrow().pending_sync_rows);
            ENGINE.with(|e| {
                let mut eng = e.borrow_mut();
                eng.record_pending_sync_rows(pending);
                if pending == 0 {
                    eng.mark_idle_if_writable();
                }
            });
            Ok(query_start)
        }
    }
}
