// In-memory packed-generation cache fragment included by `hnsw_am.rs`:
// directory cache, packed graph arrays, shared/local store dispatch, delta
// serving generations, and the serving/build telemetry counters.


/// Scan-local authoritative node locator table.
///
/// Directory pages are append-only. A higher revision wins; records with an
/// equal revision retain append order, so the last observed locator wins.
#[derive(Default)]
struct HnswDirectoryIndex {
    nodes: BTreeMap<usize, HnswDirectoryRecord>,
}

#[derive(Clone)]
struct CachedHnswDirectory {
    directory: Rc<HnswDirectoryIndex>,
}

/// Complete immutable cache identity. Segment-local node ids and generations
/// may repeat across indexes, so no subset of these fields is sufficient.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct HnswSegmentCacheKey {
    index_oid: u32,
    rel_file_number: u32,
    segment_id: u64,
    generation: u64,
}

const HNSW_CODEC_TRAINING_SAMPLE_ROWS: usize = 4_096;

fn hnsw_codec_training_sample_indices(row_count: usize) -> Vec<usize> {
    let sample_count = row_count.min(HNSW_CODEC_TRAINING_SAMPLE_ROWS);
    if sample_count == row_count {
        return (0..row_count).collect();
    }
    (0..sample_count)
        .map(|sample| sample.saturating_mul(row_count) / sample_count)
        .collect()
}

#[derive(Debug, Clone, Copy)]
struct PackedHnswNode {
    point_id: HnswPointId,
    vector_start: usize,
    layers_start: usize,
    layer_count: usize,
}

#[derive(Debug, Clone, Copy)]
struct PackedHnswLayer {
    neighbors_start: usize,
    neighbor_count: usize,
}

/// Backend-local immutable graph generation built from authoritative relation
/// pages and invalidated by the metapage publication identity.
struct PackedHnswGraph {
    dimensions: usize,
    nodes: Vec<PackedHnswNode>,
    vectors: Vec<f32>,
    layers: Vec<PackedHnswLayer>,
    neighbors: Vec<HnswNodeId>,
    quantization: Option<HnswGraphQuantization>,
}

impl PackedHnswGraph {
    fn from_records(
        records: Vec<HnswVectorRecord>,
        node_count: usize,
        dimensions: usize,
        codec_spec: Option<CodecSpec>,
    ) -> context_index::GraphResult<Self> {
        if records.len() != node_count {
            return Err(context_index::GraphError::CorruptGraph {
                message: "packed HNSW generation node count is incomplete".to_owned(),
            });
        }
        let quantization = codec_spec
            .filter(|_| !records.is_empty())
            .map(|spec| {
                let sample = hnsw_codec_training_sample_indices(records.len())
                    .into_iter()
                    .map(|index| records[index].vector.clone())
                    .collect::<Vec<_>>();
                let trained = TrainedCodecArtifact::train(spec, &sample).map_err(|error| {
                    context_index::GraphError::AdapterFailure {
                        operation: "train HNSW codec",
                        message: error.to_string(),
                    }
                })?;
                let codebook = trained.codebook().cloned().ok_or_else(|| {
                    context_index::GraphError::AdapterFailure {
                        operation: "train HNSW codec",
                        message: "quantized HNSW codec produced no codebook".to_owned(),
                    }
                })?;
                let codes = trained
                    .encode_refs(records.iter().map(|record| &record.vector))
                    .map_err(|error| context_index::GraphError::AdapterFailure {
                        operation: "encode HNSW codec rows",
                        message: error.to_string(),
                    })?
                    .ok_or_else(|| context_index::GraphError::AdapterFailure {
                        operation: "encode HNSW codec rows",
                        message: "quantized HNSW codec produced no codes".to_owned(),
                    })?;
                HnswGraphQuantization::new(
                    trained.revision(),
                    trained.reconstruction_policy(),
                    codebook,
                    codes,
                )
                .map_err(|error| context_index::GraphError::AdapterFailure {
                    operation: "publish HNSW codec artifact",
                    message: error.to_string(),
                })
            })
            .transpose()?;
        let mut nodes = Vec::with_capacity(node_count);
        let mut vectors = Vec::with_capacity(node_count.saturating_mul(dimensions));
        let mut layers = Vec::new();
        let mut neighbors = Vec::new();
        for (expected_id, record) in records.into_iter().enumerate() {
            if record.node_id.get() != expected_id || record.vector.dimension() != dimensions {
                return Err(context_index::GraphError::CorruptGraph {
                    message: "packed HNSW generation has non-contiguous nodes or dimensions"
                        .to_owned(),
                });
            }
            if record.layers.is_empty() {
                return Err(context_index::GraphError::MissingBaseLayer {
                    node_id: record.node_id,
                });
            }
            let vector_start = vectors.len();
            vectors.extend_from_slice(record.vector.as_slice());
            let layers_start = layers.len();
            for layer_neighbors in &record.layers {
                let neighbors_start = neighbors.len();
                neighbors.extend_from_slice(layer_neighbors);
                layers.push(PackedHnswLayer {
                    neighbors_start,
                    neighbor_count: layer_neighbors.len(),
                });
            }
            nodes.push(PackedHnswNode {
                // Preserve the tombstone flag for traversal so dead structural
                // nodes can be excluded from the final candidate stream.
                point_id: HnswPointId::new(record.heap_tid),
                vector_start,
                layers_start,
                layer_count: record.layers.len(),
            });
        }
        Ok(Self {
            dimensions,
            nodes,
            vectors,
            layers,
            neighbors,
            quantization,
        })
    }

    fn node(&self, node_id: HnswNodeId) -> Option<(PackedHnswNode, &[f32])> {
        let node = *self.nodes.get(node_id.get())?;
        let vector_end = node.vector_start.checked_add(self.dimensions)?;
        Some((node, self.vectors.get(node.vector_start..vector_end)?))
    }

    /// Approximate resident bytes of the packed arrays (payload only;
    /// allocator overhead is excluded, matching `estimate_index_memory`).
    fn byte_size(&self) -> u64 {
        let nodes = self.nodes.len() * size_of::<PackedHnswNode>();
        let vectors = self.vectors.len() * size_of::<f32>();
        let layers = self.layers.len() * size_of::<PackedHnswLayer>();
        let neighbors = self.neighbors.len() * size_of::<HnswNodeId>();
        let codes = self
            .quantization
            .as_ref()
            .map_or(0, |quantization| quantization.codes().as_bytes().len());
        let codebook = self.quantization.as_ref().map_or(0, |quantization| {
            quantization.codebook().resident_bytes()
        });
        (nodes + vectors + layers + neighbors + codes + codebook) as u64
    }

    /// Encodes this generation into the portable shared-image byte format
    /// (see `context_storage::packed_graph_image`), for publishing to the
    /// shared serving registry.
    fn encode_image(&self) -> Result<Vec<u8>, PackedGraphImageError> {
        let dimensions = u32::try_from(self.dimensions)
            .map_err(|_| PackedGraphImageError::CountOverflow)?;
        let nodes: Vec<PackedGraphImageNode> = self
            .nodes
            .iter()
            .map(|node| PackedGraphImageNode {
                point_id: node.point_id.get(),
                vector_start: node.vector_start as u64,
                layers_start: node.layers_start as u64,
                layer_count: node.layer_count as u64,
            })
            .collect();
        let layers: Vec<PackedGraphImageLayer> = self
            .layers
            .iter()
            .map(|layer| PackedGraphImageLayer {
                neighbors_start: layer.neighbors_start as u64,
                neighbor_count: layer.neighbor_count as u64,
            })
            .collect();
        let neighbors: Vec<u64> = self.neighbors.iter().map(|id| id.get() as u64).collect();
        encode_packed_graph_image_current(
            dimensions,
            &nodes,
            &layers,
            &neighbors,
            &self.vectors,
            self.quantization.as_ref(),
        )
    }

    fn prepare_query(
        &self,
        query: &DenseVector,
        metric: DistanceMetric,
    ) -> context_index::GraphResult<Option<PreparedQuantizedQuery>> {
        self.quantization
            .as_ref()
            .map(|quantization| quantization.codebook().prepare_query(query, metric))
            .transpose()
            .map_err(|error| context_index::GraphError::AdapterFailure {
                operation: "prepare HNSW codec query",
                message: error.to_string(),
            })
    }

    fn node_code(&self, node_id: HnswNodeId) -> Option<&[u8]> {
        self.quantization.as_ref()?.codes().code(node_id.get())
    }

    fn neighbors(&self, node: PackedHnswNode, layer: LayerIndex) -> Option<&[HnswNodeId]> {
        if layer.get() >= node.layer_count {
            return None;
        }
        let layer = self.layers.get(node.layers_start + layer.get())?;
        let end = layer.neighbors_start.checked_add(layer.neighbor_count)?;
        self.neighbors.get(layer.neighbors_start..end)
    }
}

/// Storage seam for served packed generations.
///
/// Traversal depends on these accessors rather than on the concrete backing
/// store, so the backing can move from backend-local heap memory to a shared
/// read-mostly image (the decided hybrid serving model) without touching the
/// traversal code.
#[derive(Clone)]
enum PackedGraphStore {
    Local(Arc<PackedHnswGraph>),
    /// A full-layer immutable graph mapped from an index-generation-bound file.
    Mapped(Rc<MappedPackedGraphImage>),
    /// A read view attached from the shared serving registry:
    /// another backend published this generation, so this backend serves it
    /// without rebuilding from PostgreSQL pages.
    Shared(Rc<AttachedSharedImage>),
}

/// Node identity fields shared across both `PackedGraphStore` backings.
#[derive(Debug, Clone, Copy)]
struct PackedNodeInfo {
    point_id: HnswPointId,
    layer_count: usize,
}

impl PackedGraphStore {
    fn quantization_codebook(&self) -> Option<&QuantizedCodebook> {
        match self {
            Self::Local(graph) => graph
                .quantization
                .as_ref()
                .map(HnswGraphQuantization::codebook),
            Self::Mapped(image) => image.view().quantization_codebook(),
            Self::Shared(image) => image.view().quantization_codebook(),
        }
    }

    fn node(&self, node_id: HnswNodeId) -> Option<(PackedNodeInfo, &[f32])> {
        match self {
            Self::Local(graph) => {
                let (node, vector) = graph.node(node_id)?;
                Some((
                    PackedNodeInfo {
                        point_id: node.point_id,
                        layer_count: node.layer_count,
                    },
                    vector,
                ))
            }
            Self::Mapped(image) => {
                let node = image.view().node(node_id.get())?;
                let vector = image.view().node_vector(node)?;
                let layer_count = usize::try_from(node.layer_count).ok()?;
                Some((
                    PackedNodeInfo {
                        point_id: HnswPointId::new(node.point_id),
                        layer_count,
                    },
                    vector,
                ))
            }
            Self::Shared(image) => {
                let node = image.view().node(node_id.get())?;
                let vector = image.view().node_vector(node)?;
                let layer_count = usize::try_from(node.layer_count).ok()?;
                Some((
                    PackedNodeInfo {
                        point_id: HnswPointId::new(node.point_id),
                        layer_count,
                    },
                    vector,
                ))
            }
        }
    }

    fn prepare_query(
        &self,
        query: &DenseVector,
        metric: DistanceMetric,
    ) -> context_index::GraphResult<Option<PreparedQuantizedQuery>> {
        let codebook = match self {
            Self::Local(graph) => return graph.prepare_query(query, metric),
            Self::Mapped(image) => image.view().quantization_codebook(),
            Self::Shared(image) => image.view().quantization_codebook(),
        };
        codebook
            .map(|codebook| codebook.prepare_query(query, metric))
            .transpose()
            .map_err(|error| context_index::GraphError::AdapterFailure {
                operation: "prepare packed HNSW codec query",
                message: error.to_string(),
            })
    }

    fn node_code(&self, node_id: HnswNodeId) -> Option<&[u8]> {
        match self {
            Self::Local(graph) => graph.node_code(node_id),
            Self::Mapped(image) => image.view().node_code(node_id.get()),
            Self::Shared(image) => image.view().node_code(node_id.get()),
        }
    }

    /// Looks up `node_id`'s neighbors at `layer` and appends the ones whose
    /// id is below `node_count` (excludes not-yet-published rewire targets)
    /// into `output`. Returns `Ok(false)` when `node_id` is unknown to this
    /// generation, or `Err(LayerNotFound)` when the node exists but `layer`
    /// does not.
    fn neighbors_into(
        &self,
        node_id: HnswNodeId,
        layer: LayerIndex,
        node_count: usize,
        output: &mut Vec<HnswNodeId>,
    ) -> context_index::GraphResult<bool> {
        match self {
            Self::Local(graph) => {
                let Some((node, _)) = graph.node(node_id) else {
                    output.clear();
                    return Ok(false);
                };
                let Some(neighbors) = graph.neighbors(node, layer) else {
                    return Err(context_index::GraphError::LayerNotFound { node_id, layer });
                };
                output.clear();
                output.extend(
                    neighbors
                        .iter()
                        .copied()
                        .filter(|neighbor| neighbor.get() < node_count),
                );
                Ok(true)
            }
            Self::Mapped(image) => {
                let Some(node) = image.view().node(node_id.get()) else {
                    output.clear();
                    return Ok(false);
                };
                let Some(neighbors) = image.view().neighbors(node, layer.get()) else {
                    return Err(context_index::GraphError::LayerNotFound { node_id, layer });
                };
                output.clear();
                output.extend(
                    neighbors
                        .filter_map(|id| usize::try_from(id).ok().map(HnswNodeId::new))
                        .filter(|neighbor| neighbor.get() < node_count),
                );
                Ok(true)
            }
            Self::Shared(image) => {
                let Some(node) = image.view().node(node_id.get()) else {
                    output.clear();
                    return Ok(false);
                };
                let Some(neighbors) = image.view().neighbors(node, layer.get()) else {
                    return Err(context_index::GraphError::LayerNotFound { node_id, layer });
                };
                output.clear();
                output.extend(
                    neighbors
                        .filter_map(|id| usize::try_from(id).ok().map(HnswNodeId::new))
                        .filter(|neighbor| neighbor.get() < node_count),
                );
                Ok(true)
            }
        }
    }

}

/// A packed generation ready to serve traversal: a base, either
/// backend-local or attached from the shared registry.
///
/// A generation is immutable and whole. It used to carry an optional overlay
/// of nodes that had changed since `revision_watermark`, so a stale base could
/// be patched instead of repacked; that was retired with the segmented write
/// path, which absorbs writes on disk instead. Patching was only ever correct
/// because a compaction dirties every node and so happened to shadow the
/// entire stale base -- an emergent property of the layout, not an invariant,
/// and one that a partially-dirty generation would have broken by mixing two
/// node numberings.
#[derive(Clone)]
struct HnswPackedGeneration {
    base: PackedGraphStore,
}

impl HnswPackedGeneration {
    fn node(&self, node_id: HnswNodeId) -> Option<(PackedNodeInfo, &[f32])> {
        self.base.node(node_id)
    }

    fn neighbors_into(
        &self,
        node_id: HnswNodeId,
        layer: LayerIndex,
        node_count: usize,
        output: &mut Vec<HnswNodeId>,
    ) -> context_index::GraphResult<bool> {
        self.base.neighbors_into(node_id, layer, node_count, output)
    }

    fn prepare_query(
        &self,
        query: &DenseVector,
        metric: DistanceMetric,
    ) -> context_index::GraphResult<Option<PreparedQuantizedQuery>> {
        self.base.prepare_query(query, metric)
    }

    fn node_code(&self, node_id: HnswNodeId) -> Option<&[u8]> {
        self.base.node_code(node_id)
    }

    fn matches_codec(&self, meta: HnswMetaPage) -> bool {
        meta.accepts_codec_codebook(self.base.quantization_codebook())
    }

    fn parallel_local_graph(&self) -> Option<Arc<PackedHnswGraph>> {
        match &self.base {
            PackedGraphStore::Local(graph) => Some(Arc::clone(graph)),
            PackedGraphStore::Mapped(_) | PackedGraphStore::Shared(_) => None,
        }
    }
}

#[derive(Clone)]
struct CachedPackedHnswGraph {
    /// Physical relation file this pack was built from. REINDEX swaps the
    /// relfilenode while keeping the index OID, and the fresh build's
    /// directory revisions restart low enough that the stale-patch path's
    /// `dirty_since` watermark sees no drift — so without this field a
    /// cached pack of the pre-REINDEX graph is re-served as current.
    rel_file_number: u32,
    graph: HnswPackedGeneration,
}

/// Backend-local packed-generation serving telemetry.
///
/// The packed cache is currently backend-local, so these counters describe
/// the current backend only; they exist so operators can observe pack
/// builds, reuses, and their cost without guessing from latency cliffs.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct HnswServingStats {
    pub(crate) pack_builds: u64,
    pub(crate) pack_reuses: u64,
    pub(crate) last_pack_bytes: u64,
    pub(crate) last_pack_millis: u64,
    pub(crate) total_pack_millis: u64,
    /// Packed generations attached from the shared registry
    /// instead of rebuilt from PostgreSQL pages.
    pub(crate) shared_attaches: u64,
    /// Backend-local builds successfully published to the shared registry.
    pub(crate) shared_publishes: u64,
    /// Backend-local builds whose publish was skipped (GUC off, over
    /// budget, or segment creation failed) — service continued locally.
    pub(crate) shared_publish_skips: u64,
    /// Generations attached from immutable identity-bound mapped files.
    pub(crate) mapped_attaches: u64,
    /// Backend-local packs successfully published as mapped files.
    pub(crate) mapped_publishes: u64,
    /// Mapped publication attempts skipped or failed safely.
    pub(crate) mapped_publish_skips: u64,
    /// Queries served from unpacked directory reads because no pack was
    /// available and `pgcontext.hnsw_pack_on_first_use` was off.
    pub(crate) page_native_fallbacks: u64,
    /// Rows appended to a segmented-write delta region (P2-S3) instead of
    /// being spliced into the HNSW graph, including tombstones appended by
    /// VACUUM for delta-only rows.
    pub(crate) delta_segment_records: u64,
    /// Scans that merged an exact delta-region scan with base-graph
    /// candidates because the index's delta region held at least one record.
    pub(crate) delta_segment_scans: u64,
    /// Full active deltas frozen as immutable graph segments.
    pub(crate) segment_rotations: u64,
    /// Queries that searched more than one immutable graph segment.
    pub(crate) multi_segment_scans: u64,
    /// Multi-segment scans admitted to the bounded pure worker pool.
    pub(crate) parallel_segment_scans: u64,
    /// Multi-segment scans kept serial because PostgreSQL relation access is
    /// backend-affine and no admitted parallel worker was available.
    pub(crate) serial_segment_degradations: u64,
    /// Largest immutable segment fan-out observed by this backend.
    pub(crate) max_segments_observed: u64,
    pub(crate) segment_compactions: u64,
    pub(crate) parallel_admission_denials: u64,
    pub(crate) segment_merge_candidates: u64,
    pub(crate) frozen_mutation_records_scanned: u64,
    pub(crate) compaction_input_rows: u64,
    pub(crate) compaction_output_rows: u64,
    pub(crate) compaction_wal_pages: u64,
    pub(crate) compaction_peak_projected_bytes: u64,
}

thread_local! {
    static HNSW_DIRECTORY_CACHE: RefCell<BTreeMap<HnswSegmentCacheKey, CachedHnswDirectory>> =
        const { RefCell::new(BTreeMap::new()) };
    static HNSW_PACKED_GRAPH_CACHE: RefCell<BTreeMap<HnswSegmentCacheKey, CachedPackedHnswGraph>> =
        const { RefCell::new(BTreeMap::new()) };
    static HNSW_SERVING_STATS: Cell<HnswServingStats> =
        const { Cell::new(HnswServingStats {
            pack_builds: 0,
            pack_reuses: 0,
            last_pack_bytes: 0,
            last_pack_millis: 0,
            total_pack_millis: 0,
            shared_attaches: 0,
            shared_publishes: 0,
            shared_publish_skips: 0,
            mapped_attaches: 0,
            mapped_publishes: 0,
            mapped_publish_skips: 0,
            page_native_fallbacks: 0,
            delta_segment_records: 0,
            delta_segment_scans: 0,
            segment_rotations: 0,
            multi_segment_scans: 0,
            parallel_segment_scans: 0,
            serial_segment_degradations: 0,
            max_segments_observed: 0,
            segment_compactions: 0,
            parallel_admission_denials: 0,
            segment_merge_candidates: 0,
            frozen_mutation_records_scanned: 0,
            compaction_input_rows: 0,
            compaction_output_rows: 0,
            compaction_wal_pages: 0,
            compaction_peak_projected_bytes: 0,
        }) };
}

/// Phase timing of the most recent HNSW bulk build in this backend.
///
/// `graph_millis` covers the heap scan plus in-memory graph construction;
/// `write_millis` covers snapshot extraction, page writes, and Generic-WAL
/// emission. The split directs build-performance work at the dominant phase.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct HnswBuildProfile {
    pub(crate) tuples: u64,
    pub(crate) graph_millis: u64,
    pub(crate) write_millis: u64,
}

thread_local! {
    static HNSW_BUILD_PROFILE: Cell<HnswBuildProfile> =
        const { Cell::new(HnswBuildProfile {
            tuples: 0,
            graph_millis: 0,
            write_millis: 0,
        }) };
}

/// Returns the phase profile of this backend's most recent HNSW bulk build.
pub(crate) fn hnsw_build_profile_snapshot() -> HnswBuildProfile {
    HNSW_BUILD_PROFILE.with(Cell::get)
}

fn record_hnsw_build_profile(profile: HnswBuildProfile) {
    HNSW_BUILD_PROFILE.with(|slot| slot.set(profile));
}

fn saturating_elapsed_millis(started: std::time::Instant) -> u64 {
    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
}

/// Returns this backend's packed-generation serving counters.
pub(crate) fn hnsw_serving_stats_snapshot() -> HnswServingStats {
    HNSW_SERVING_STATS.with(Cell::get)
}

fn record_hnsw_pack_reuse() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.pack_reuses = current.pack_reuses.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_pack_build(bytes: u64, millis: u64) {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.pack_builds = current.pack_builds.saturating_add(1);
        current.last_pack_bytes = bytes;
        current.last_pack_millis = millis;
        current.total_pack_millis = current.total_pack_millis.saturating_add(millis);
        stats.set(current);
    });
}

/// Records the complete packed footprint concurrently resident for the most
/// recent multi-segment serving generation.
fn record_hnsw_generation_bytes(bytes: u64) {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.last_pack_bytes = bytes;
        stats.set(current);
    });
}

fn record_hnsw_shared_attach() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.shared_attaches = current.shared_attaches.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_mapped_attach() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.mapped_attaches = current.mapped_attaches.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_mapped_publish(published: bool) {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        if published {
            current.mapped_publishes = current.mapped_publishes.saturating_add(1);
        } else {
            current.mapped_publish_skips = current.mapped_publish_skips.saturating_add(1);
        }
        stats.set(current);
    });
}

fn record_hnsw_shared_publish(published: bool) {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        if published {
            current.shared_publishes = current.shared_publishes.saturating_add(1);
        } else {
            current.shared_publish_skips = current.shared_publish_skips.saturating_add(1);
        }
        stats.set(current);
    });
}



fn record_hnsw_page_native_fallback() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.page_native_fallbacks = current.page_native_fallbacks.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_delta_segment_record() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.delta_segment_records = current.delta_segment_records.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_delta_segment_scan() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.delta_segment_scans = current.delta_segment_scans.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_segment_rotation() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.segment_rotations = current.segment_rotations.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_multi_segment_scan(segment_count: usize) {
    if segment_count <= 1 {
        return;
    }
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.multi_segment_scans = current.multi_segment_scans.saturating_add(1);
        current.max_segments_observed = current
            .max_segments_observed
            .max(u64::try_from(segment_count).unwrap_or(u64::MAX));
        stats.set(current);
    });
}

fn record_hnsw_parallel_segment_scan() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.parallel_segment_scans = current.parallel_segment_scans.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_serial_segment_degradation() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.serial_segment_degradations =
            current.serial_segment_degradations.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_parallel_admission_denial() {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.parallel_admission_denials = current.parallel_admission_denials.saturating_add(1);
        stats.set(current);
    });
}

fn record_hnsw_segment_merge(base_candidates: usize, frozen_mutations: usize) {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.segment_merge_candidates = current
            .segment_merge_candidates
            .saturating_add(u64::try_from(base_candidates).unwrap_or(u64::MAX));
        current.frozen_mutation_records_scanned = current
            .frozen_mutation_records_scanned
            .saturating_add(u64::try_from(frozen_mutations).unwrap_or(u64::MAX));
        stats.set(current);
    });
}

fn record_hnsw_segment_compaction(
    input_rows: u64,
    output_rows: usize,
    wal_pages: u64,
    projected_bytes: usize,
) {
    HNSW_SERVING_STATS.with(|stats| {
        let mut current = stats.get();
        current.segment_compactions = current.segment_compactions.saturating_add(1);
        current.compaction_input_rows = current.compaction_input_rows.saturating_add(input_rows);
        current.compaction_output_rows = current
            .compaction_output_rows
            .saturating_add(u64::try_from(output_rows).unwrap_or(u64::MAX));
        current.compaction_wal_pages = current.compaction_wal_pages.saturating_add(wal_pages);
        current.compaction_peak_projected_bytes = current
            .compaction_peak_projected_bytes
            .max(u64::try_from(projected_bytes).unwrap_or(u64::MAX));
        stats.set(current);
    });
}

fn retire_hnsw_segment_caches(
    database_oid: u32,
    index_oid: u32,
    rel_file_number: u32,
    segment_ids: &[u64],
) {
    HNSW_DIRECTORY_CACHE.with(|cache| {
        cache.borrow_mut().retain(|key, _| {
            key.index_oid != index_oid
                || key.rel_file_number != rel_file_number
                || !segment_ids.contains(&key.segment_id)
        });
    });
    HNSW_PACKED_GRAPH_CACHE.with(|cache| {
        cache.borrow_mut().retain(|key, _| {
            key.index_oid != index_oid
                || key.rel_file_number != rel_file_number
                || !segment_ids.contains(&key.segment_id)
        });
    });
    for segment_id in segment_ids {
        evict_shared_image(
            database_oid,
            index_oid,
            rel_file_number,
            *segment_id,
        );
    }
    retire_mapped_segments(
        database_oid,
        index_oid,
        rel_file_number,
        segment_ids,
    );
}

impl HnswDirectoryIndex {
    fn observe(&mut self, record: HnswDirectoryRecord) {
        if record.key_kind != GraphDirectoryKeyKind::Node || record.ordinal != 0 {
            return;
        }
        let Ok(identity) = usize::try_from(record.identity) else {
            return;
        };
        let replace = self
            .nodes
            .get(&identity)
            .is_none_or(|current| record.revision >= current.revision);
        if replace {
            self.nodes.insert(identity, record);
        }
    }

    fn node(&self, node_id: HnswNodeId) -> Option<HnswDirectoryRecord> {
        self.nodes.get(&node_id.get()).copied()
    }


}
