kglite 0.16.5

// src/graph/storage/interner.rs
//
// InternedKey + StringInterner + serde thread-local guards.
//
// InternedKey is a compact FNV-1a hash of a property/type-name string;
// StringInterner holds the reverse mapping. Serde round-trips keys
// through their original strings using thread-local interner pointers
// installed by the RAII guards (`SerdeSerializeGuard` /
// `SerdeDeserializeGuard`). A third guard (`StripPropertiesGuard`)
// enables the v3 topology-mode serialization path.
//
// Extracted from `src/graph/schema.rs` in Phase 7 (Stage 2.2).

use rustc_hash::FxHashMap;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::cell::Cell;
use std::hash::{Hash, Hasher};
use std::sync::Arc;

use crate::graph::cow::cow_mut;

/// Two distinct strings attempted to claim the same persisted u64 identity.
/// The first mapping remains unchanged when this error is returned.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InternerCollision {
    pub key: u64,
    pub existing: String,
    pub conflicting: String,
}

impl std::fmt::Display for InternerCollision {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "InternedKey hash collision at {}: '{}' conflicts with '{}'",
            self.key, self.conflicting, self.existing
        )
    }
}

impl std::error::Error for InternerCollision {}

/// A compact property key backed by a hash of the original string.
/// Lookups via `get_property(key)` compute the hash inline — no interner needed.
/// Only methods that output string keys (e.g. `property_iter`) require the interner.
#[repr(transparent)]
#[derive(Debug, Default, Copy, Clone, Eq, Ord, PartialEq, PartialOrd)]
pub struct InternedKey(u64);

impl InternedKey {
    /// Compute the interned key from a string. **Must be deterministic across
    /// processes and library versions** — `DiskNodeSlot.node_type` persists
    /// this as raw u64 on disk (`disk_graph.rs`), and the loader resolves it
    /// via the freshly-built interner's hashes. A per-process random seed
    /// (e.g. `DefaultHasher`) would break cross-process disk loads.
    ///
    /// Uses FNV-1a 64-bit. Fast, zero-alloc, dependency-free, deterministic.
    /// Our corpus (property names, type names) is at most a few thousand
    /// short strings, so collision risk is negligible; `StringInterner`
    /// nevertheless detects and rejects conflicting strings in every build.
    #[inline]
    // This established persisted-key constructor predates FromStr and is part of the public API.
    #[allow(clippy::should_implement_trait)]
    pub fn from_str(s: &str) -> Self {
        const FNV_OFFSET: u64 = 0xcbf29ce484222325;
        const FNV_PRIME: u64 = 0x100000001b3;
        let mut h = FNV_OFFSET;
        for &byte in s.as_bytes() {
            h ^= byte as u64;
            h = h.wrapping_mul(FNV_PRIME);
        }
        InternedKey(h)
    }

    /// Get the raw u64 hash value. Used for disk storage.
    #[inline]
    pub fn as_u64(&self) -> u64 {
        self.0
    }

    /// Reconstruct from a raw u64 hash value. Used when loading from disk.
    #[inline]
    pub fn from_u64(v: u64) -> Self {
        InternedKey(v)
    }
}

impl Hash for InternedKey {
    #[inline]
    fn hash<H: Hasher>(&self, state: &mut H) {
        state.write_u64(self.0);
    }
}

/// Serializes InternedKey as its original string (backward-compatible with
/// HashMap<String, Value> on disk). Requires the thread-local SERIALIZE_INTERNER
/// to be set before the top-level serialize call.
impl Serialize for InternedKey {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        SERIALIZE_INTERNER.with(|cell| {
            let ptr = cell
                .get()
                .expect("BUG: SERIALIZE_INTERNER not set during InternedKey serialization");
            // SAFETY: ptr is set by SerdeInternerGuard which ensures the reference
            // outlives the serialize call (the guard lives on the caller's stack).
            let interner = unsafe { &*ptr };
            interner.resolve(*self).serialize(serializer)
        })
    }
}

/// Deserializes InternedKey from a string (backward-compatible with
/// HashMap<String, Value> on disk). Registers the string in the thread-local
/// DESERIALIZE_INTERNER if set.
///
/// Uses a custom Visitor to avoid String allocation: the binary Serde reader
/// provides borrowed &str directly from the decompressed buffer. Only the
/// first occurrence of each key allocates (in the interner). For ~5.6M
/// property keys with ~200 unique ones, this eliminates ~5.6M allocations.
impl<'de> Deserialize<'de> for InternedKey {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        struct KeyVisitor;
        impl<'de> serde::de::Visitor<'de> for KeyVisitor {
            type Value = InternedKey;
            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
                f.write_str("a string key")
            }
            /// Fast path: hash borrowed &str directly, no String allocation.
            fn visit_str<E: serde::de::Error>(self, v: &str) -> Result<Self::Value, E> {
                let key = InternedKey::from_str(v);
                DESERIALIZE_INTERNER.with(|cell| -> Result<(), E> {
                    if let Some(ptr) = cell.get() {
                        // SAFETY: ptr is set by SerdeInternerGuard which
                        // ensures the &mut reference outlives the
                        // deserialize call (the guard lives on the
                        // caller's stack, same pattern as Serialize above).
                        let interner = unsafe { &mut *ptr };
                        interner.try_register(key, v).map_err(E::custom)?;
                    }
                    Ok(())
                })?;
                Ok(key)
            }
            /// Fallback for formats that provide owned Strings (e.g. JSON).
            fn visit_string<E: serde::de::Error>(self, v: String) -> Result<Self::Value, E> {
                self.visit_str(&v)
            }
        }
        deserializer.deserialize_str(KeyVisitor)
    }
}

/// Reverse mapping from InternedKey → original string.
/// Used for serialization and for methods that output string keys.
#[derive(Debug, Clone, Default)]
pub struct StringInterner {
    /// FxHash, not SipHash: `InternedKey` is already an FNV `u64`, so the std
    /// cryptographic hasher is pure overhead. `try_resolve`/`resolve` run per
    /// row (e.g. `node_type_str` in `resolve_node_property`) — see the
    /// 2026-05-29 samply profile (SipHash ~23% of in-memory query CPU).
    /// `Arc`-shared and copy-on-write. `StringInterner` is one of the fields
    /// the rollback shell clones before every mutating statement
    /// (`dir_graph::rollback`), and this table is O(distinct names) — measured
    /// at ~41 ns per entry to copy, which on any real schema is the whole
    /// remaining per-statement shell cost once the six `DirGraph` maps are
    /// shared. `try_register` below reads before it forks, so a statement that
    /// interns no *new* name copies nothing.
    strings: Arc<FxHashMap<InternedKey, String>>,
}

impl StringInterner {
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a key-string mapping without changing the first mapping on a
    /// collision. Detection is active in debug and release builds.
    #[inline]
    pub fn try_register(&mut self, key: InternedKey, s: &str) -> Result<(), InternerCollision> {
        if let Some(existing) = self.strings.get(&key) {
            if &**existing != s {
                return Err(InternerCollision {
                    key: key.as_u64(),
                    existing: existing.to_string(),
                    conflicting: s.to_string(),
                });
            }
            return Ok(());
        }
        cow_mut(&mut self.strings).insert(key, s.to_string());
        Ok(())
    }

    /// Fallible public interning path for user- or file-derived names.
    #[inline]
    pub fn try_get_or_intern(&mut self, s: &str) -> Result<InternedKey, InternerCollision> {
        let key = InternedKey::from_str(s);
        self.try_register(key, s)?;
        Ok(key)
    }

    /// Internal infallible path. Callers must have prevalidated user/file names
    /// or use fixed literals. A missed collision panics in every build rather
    /// than silently aliasing data in release.
    #[inline]
    pub(crate) fn get_or_intern(&mut self, s: &str) -> InternedKey {
        self.try_get_or_intern(s)
            .unwrap_or_else(|collision| panic!("{collision}"))
    }

    /// Validate names against the current interner and against one another
    /// without mutating `self`. Returns their keys in input order.
    pub fn validate_names<'a>(
        &self,
        names: impl IntoIterator<Item = &'a str>,
    ) -> Result<Vec<InternedKey>, InternerCollision> {
        let names = names.into_iter();
        let (lower, _) = names.size_hint();
        let mut keys = Vec::with_capacity(lower);
        let mut staged: FxHashMap<InternedKey, &'a str> = FxHashMap::default();

        for name in names {
            let key = InternedKey::from_str(name);
            if let Some(existing) = self.strings.get(&key).map(|s| &**s) {
                if existing != name {
                    return Err(InternerCollision {
                        key: key.as_u64(),
                        existing: existing.to_string(),
                        conflicting: name.to_string(),
                    });
                }
            } else if let Some(existing) = staged.get(&key).copied() {
                if existing != name {
                    return Err(InternerCollision {
                        key: key.as_u64(),
                        existing: existing.to_string(),
                        conflicting: name.to_string(),
                    });
                }
            } else {
                staged.insert(key, name);
            }
            keys.push(key);
        }

        Ok(keys)
    }

    /// Resolve an InternedKey back to its string. Panics if the key is unknown.
    #[inline]
    pub fn resolve(&self, key: InternedKey) -> &str {
        self.strings.get(&key).map(|s| &**s).unwrap_or_else(|| {
            panic!(
                "InternedKey {} not found in StringInterner ({} entries)",
                key.as_u64(),
                self.strings.len()
            )
        })
    }

    /// Iterate over all (key, string) pairs in the interner.
    pub fn iter(&self) -> impl Iterator<Item = (InternedKey, &str)> {
        self.strings.iter().map(|(&k, v)| (k, &**v))
    }

    /// Resolve an InternedKey back to its string, returning None if unknown.
    #[inline]
    pub fn try_resolve(&self, key: InternedKey) -> Option<&str> {
        self.strings.get(&key).map(|s| &**s)
    }

    /// Compute the InternedKey for a string (if it exists in the interner).
    #[inline]
    pub fn try_resolve_to_key(&self, s: &str) -> Option<InternedKey> {
        let key = InternedKey::from_str(s);
        if self
            .strings
            .get(&key)
            .is_some_and(|existing| &**existing == s)
        {
            Some(key)
        } else {
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn raw_collision_is_typed_and_keeps_first_mapping() {
        let mut interner = StringInterner::new();
        let key = InternedKey::from_str("second");
        interner.try_register(key, "first").unwrap();
        let err = interner.try_register(key, "second").unwrap_err();
        assert_eq!(err.key, key.as_u64());
        assert_eq!(err.existing, "first");
        assert_eq!(err.conflicting, "second");
        assert_eq!(interner.resolve(key), "first");
        assert_eq!(interner.try_resolve_to_key("second"), None);
    }

    #[test]
    fn batch_validation_detects_internal_collision_without_mutation() {
        let mut interner = StringInterner::new();
        let key = InternedKey::from_str("incoming");
        interner
            .try_register(key, "existing-with-forced-key")
            .unwrap();
        let before: Vec<_> = interner.iter().map(|(k, v)| (k, v.to_string())).collect();
        assert!(interner.validate_names(["ordinary", "incoming"]).is_err());
        let after: Vec<_> = interner.iter().map(|(k, v)| (k, v.to_string())).collect();
        assert_eq!(after, before);
    }

    #[test]
    fn serde_collision_is_a_deserializer_error() {
        let incoming = "persisted-name";
        let mut interner = StringInterner::new();
        interner
            .try_register(InternedKey::from_str(incoming), "conflicting-existing")
            .unwrap();
        let bytes = crate::serde_codec::encode_versioned(
            crate::serde_codec::CURRENT_CODEC,
            incoming,
            u64::MAX,
        )
        .unwrap();
        let guard = SerdeDeserializeGuard::new(&mut interner);
        let decoded = crate::serde_codec::decode_exact_with::<InternedKey>(
            crate::serde_codec::CURRENT_CODEC,
            &bytes,
            bytes.len() as u64,
            crate::serde_codec::DecodeLimits::new(u64::MAX, u64::MAX),
        );
        drop(guard);
        assert!(decoded.is_err());
        assert_eq!(
            interner.resolve(InternedKey::from_str(incoming)),
            "conflicting-existing"
        );
    }
}

// ─── Thread-local serde support ───────────────────────────────────────────────

thread_local! {
    static SERIALIZE_INTERNER: Cell<Option<*const StringInterner>> = const { Cell::new(None) };
    static DESERIALIZE_INTERNER: Cell<Option<*mut StringInterner>> = const { Cell::new(None) };
    /// When true, PropertyStorage::Serialize emits an empty map (v3 topology mode).
    pub(crate) static STRIP_PROPERTIES: Cell<bool> = const { Cell::new(false) };
}

/// RAII guard that sets the thread-local interner for serialization.
/// The interner reference must outlive the guard (enforced by the lifetime).
pub(crate) struct SerdeSerializeGuard<'a> {
    _phantom: std::marker::PhantomData<&'a StringInterner>,
}

impl<'a> SerdeSerializeGuard<'a> {
    pub fn new(interner: &'a StringInterner) -> Self {
        SERIALIZE_INTERNER.with(|cell| cell.set(Some(interner as *const StringInterner)));
        SerdeSerializeGuard {
            _phantom: std::marker::PhantomData,
        }
    }
}

impl Drop for SerdeSerializeGuard<'_> {
    fn drop(&mut self) {
        SERIALIZE_INTERNER.with(|cell| cell.set(None));
    }
}

/// RAII guard that sets the thread-local interner for deserialization.
pub(crate) struct SerdeDeserializeGuard<'a> {
    _phantom: std::marker::PhantomData<&'a mut StringInterner>,
}

impl<'a> SerdeDeserializeGuard<'a> {
    pub fn new(interner: &'a mut StringInterner) -> Self {
        DESERIALIZE_INTERNER.with(|cell| cell.set(Some(interner as *mut StringInterner)));
        SerdeDeserializeGuard {
            _phantom: std::marker::PhantomData,
        }
    }
}

impl Drop for SerdeDeserializeGuard<'_> {
    fn drop(&mut self) {
        DESERIALIZE_INTERNER.with(|cell| cell.set(None));
    }
}

/// RAII guard that enables property stripping during serialization.
/// While active, PropertyStorage::Serialize emits empty maps (v3 topology mode).
pub(crate) struct StripPropertiesGuard;

impl StripPropertiesGuard {
    pub fn new() -> Self {
        STRIP_PROPERTIES.with(|cell| cell.set(true));
        StripPropertiesGuard
    }
}

impl Drop for StripPropertiesGuard {
    fn drop(&mut self) {
        STRIP_PROPERTIES.with(|cell| cell.set(false));
    }
}