kitedb 0.2.15

//! Kite - High-level API for KiteDB
//!
//! The Kite struct provides a clean, ergonomic API for graph operations.
//! It wraps the lower-level SingleFileDB with schema definitions and type-safe operations.
//!
//! Provides:
//! - Node and edge CRUD operations
//! - Graph traversal with fluent API
//! - Shortest path finding (Dijkstra, BFS, Yen's k-shortest)
//! - Schema-based type safety
//!
//! Ported from src/api/kite.ts

use crate::core::single_file::{
  close_single_file, is_single_file_path, open_single_file, single_file_extension, FullEdge,
  SingleFileDB, SingleFileOpenOptions, SyncMode,
};
use crate::error::{KiteError, Result};
use crate::types::*;

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;

// ============================================================================
// Single-file transaction wrappers
// ============================================================================

#[derive(Debug, Clone, Default)]
struct NodeOpts {
  key: Option<String>,
  labels: Option<Vec<LabelId>>,
  props: Option<Vec<(PropKeyId, PropValue)>>,
}

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

  fn with_key(mut self, key: impl Into<String>) -> Self {
    self.key = Some(key.into());
    self
  }

  #[allow(dead_code)]
  fn with_label(mut self, label: LabelId) -> Self {
    self.labels.get_or_insert_with(Vec::new).push(label);
    self
  }

  #[allow(dead_code)]
  fn with_prop(mut self, key: PropKeyId, value: PropValue) -> Self {
    self.props.get_or_insert_with(Vec::new).push((key, value));
    self
  }
}

struct TxHandle<'a> {
  db: &'a SingleFileDB,
  finished: bool,
  owns_tx: bool,
}

impl<'a> TxHandle<'a> {
  fn new(db: &'a SingleFileDB, owns_tx: bool) -> Self {
    Self {
      db,
      finished: false,
      owns_tx,
    }
  }
}

impl<'a> Drop for TxHandle<'a> {
  fn drop(&mut self) {
    if !self.finished {
      if self.owns_tx {
        let _ = self.db.rollback();
      }
      self.finished = true;
    }
  }
}

fn begin_tx(db: &SingleFileDB) -> Result<TxHandle<'_>> {
  if db.has_transaction() {
    db.require_write_tx()?;
    return Ok(TxHandle::new(db, false));
  }

  db.begin(false)?;
  Ok(TxHandle::new(db, true))
}

fn commit(handle: &mut TxHandle) -> Result<()> {
  if handle.owns_tx {
    handle.db.commit()?;
  }
  handle.finished = true;
  Ok(())
}

fn rollback(handle: &mut TxHandle) -> Result<()> {
  if handle.owns_tx {
    handle.db.rollback()?;
  }
  handle.finished = true;
  Ok(())
}

#[derive(Debug, Clone, Default)]
struct ListEdgesOptions {
  pub etype: Option<ETypeId>,
}

fn create_node(handle: &mut TxHandle, opts: NodeOpts) -> Result<NodeId> {
  let node_id = handle.db.create_node(opts.key.as_deref())?;
  if let Some(labels) = opts.labels {
    for label_id in labels {
      handle.db.add_node_label(node_id, label_id)?;
    }
  }
  if let Some(props) = opts.props {
    for (key_id, value) in props {
      handle.db.set_node_prop(node_id, key_id, value)?;
    }
  }
  Ok(node_id)
}

fn create_node_with_id(handle: &mut TxHandle, node_id: NodeId, opts: NodeOpts) -> Result<NodeId> {
  let node_id = handle
    .db
    .create_node_with_id(node_id, opts.key.as_deref())?;
  if let Some(labels) = opts.labels {
    for label_id in labels {
      handle.db.add_node_label(node_id, label_id)?;
    }
  }
  if let Some(props) = opts.props {
    for (key_id, value) in props {
      handle.db.set_node_prop(node_id, key_id, value)?;
    }
  }
  Ok(node_id)
}

fn delete_node(handle: &mut TxHandle, node_id: NodeId) -> Result<bool> {
  if !handle.db.node_exists(node_id) {
    return Ok(false);
  }
  handle.db.delete_node(node_id)?;
  Ok(true)
}

fn node_exists(handle: &TxHandle, node_id: NodeId) -> bool {
  handle.db.node_exists(node_id)
}

fn node_exists_db(db: &SingleFileDB, node_id: NodeId) -> bool {
  db.node_exists(node_id)
}

fn node_by_key(handle: &TxHandle, key: &str) -> Option<NodeId> {
  handle.db.node_by_key(key)
}

fn node_by_key_db(db: &SingleFileDB, key: &str) -> Option<NodeId> {
  db.node_by_key(key)
}

fn node_prop(handle: &TxHandle, node_id: NodeId, key_id: PropKeyId) -> Option<PropValue> {
  handle.db.node_prop(node_id, key_id)
}

fn node_prop_db(db: &SingleFileDB, node_id: NodeId, key_id: PropKeyId) -> Option<PropValue> {
  db.node_prop(node_id, key_id)
}

fn set_node_prop(
  handle: &mut TxHandle,
  node_id: NodeId,
  key_id: PropKeyId,
  value: PropValue,
) -> Result<()> {
  handle.db.set_node_prop(node_id, key_id, value)
}

fn del_node_prop(handle: &mut TxHandle, node_id: NodeId, key_id: PropKeyId) -> Result<()> {
  handle.db.delete_node_prop(node_id, key_id)
}

fn upsert_node_with_props<I>(handle: &mut TxHandle, key: &str, props: I) -> Result<(NodeId, bool)>
where
  I: IntoIterator<Item = (PropKeyId, Option<PropValue>)>,
{
  let (node_id, created) = match handle.db.node_by_key(key) {
    Some(existing) => (existing, false),
    None => (create_node(handle, NodeOpts::new().with_key(key))?, true),
  };

  for (key_id, value_opt) in props {
    match value_opt {
      Some(value) => set_node_prop(handle, node_id, key_id, value)?,
      None => del_node_prop(handle, node_id, key_id)?,
    }
  }

  Ok((node_id, created))
}

fn upsert_node_by_id_with_props<I>(
  handle: &mut TxHandle,
  node_id: NodeId,
  opts: NodeOpts,
  props: I,
) -> Result<(NodeId, bool)>
where
  I: IntoIterator<Item = (PropKeyId, Option<PropValue>)>,
{
  let created = if handle.db.node_exists(node_id) {
    false
  } else {
    create_node_with_id(handle, node_id, opts)?;
    true
  };

  for (key_id, value_opt) in props {
    match value_opt {
      Some(value) => set_node_prop(handle, node_id, key_id, value)?,
      None => del_node_prop(handle, node_id, key_id)?,
    }
  }

  Ok((node_id, created))
}

fn add_edge(handle: &mut TxHandle, src: NodeId, etype: ETypeId, dst: NodeId) -> Result<()> {
  handle.db.add_edge(src, etype, dst)
}

fn delete_edge(handle: &mut TxHandle, src: NodeId, etype: ETypeId, dst: NodeId) -> Result<bool> {
  if !handle.db.edge_exists(src, etype, dst) {
    return Ok(false);
  }
  handle.db.delete_edge(src, etype, dst)?;
  Ok(true)
}

fn edge_exists(handle: &TxHandle, src: NodeId, etype: ETypeId, dst: NodeId) -> bool {
  handle.db.edge_exists(src, etype, dst)
}

fn edge_exists_db(db: &SingleFileDB, src: NodeId, etype: ETypeId, dst: NodeId) -> bool {
  db.edge_exists(src, etype, dst)
}

fn neighbors_out_db(db: &SingleFileDB, node_id: NodeId, etype: Option<ETypeId>) -> Vec<NodeId> {
  match etype {
    Some(filter) => db.out_neighbors(node_id, filter),
    None => db
      .out_edges(node_id)
      .into_iter()
      .map(|(_, dst)| dst)
      .collect(),
  }
}

fn neighbors_in_db(db: &SingleFileDB, node_id: NodeId, etype: Option<ETypeId>) -> Vec<NodeId> {
  match etype {
    Some(filter) => db.in_neighbors(node_id, filter),
    None => db
      .in_edges(node_id)
      .into_iter()
      .map(|(_, src)| src)
      .collect(),
  }
}

fn edge_prop_db(
  db: &SingleFileDB,
  src: NodeId,
  etype: ETypeId,
  dst: NodeId,
  key_id: PropKeyId,
) -> Option<PropValue> {
  db.edge_prop(src, etype, dst, key_id)
}

fn edge_props_db(
  db: &SingleFileDB,
  src: NodeId,
  etype: ETypeId,
  dst: NodeId,
) -> Option<HashMap<PropKeyId, PropValue>> {
  db.edge_props(src, etype, dst)
}

fn set_edge_prop(
  handle: &mut TxHandle,
  src: NodeId,
  etype: ETypeId,
  dst: NodeId,
  key_id: PropKeyId,
  value: PropValue,
) -> Result<()> {
  handle.db.set_edge_prop(src, etype, dst, key_id, value)
}

fn del_edge_prop(
  handle: &mut TxHandle,
  src: NodeId,
  etype: ETypeId,
  dst: NodeId,
  key_id: PropKeyId,
) -> Result<()> {
  handle.db.delete_edge_prop(src, etype, dst, key_id)
}

fn upsert_edge_with_props<I>(
  handle: &mut TxHandle,
  src: NodeId,
  etype: ETypeId,
  dst: NodeId,
  props: I,
) -> Result<bool>
where
  I: IntoIterator<Item = (PropKeyId, Option<PropValue>)>,
{
  handle.db.upsert_edge_with_props(src, etype, dst, props)
}

fn list_nodes(db: &SingleFileDB) -> Vec<NodeId> {
  db.list_nodes()
}

fn list_edges(db: &SingleFileDB, options: ListEdgesOptions) -> Vec<FullEdge> {
  db.list_edges(options.etype)
}

fn count_nodes(db: &SingleFileDB) -> u64 {
  db.count_nodes() as u64
}

fn count_edges(db: &SingleFileDB, etype_filter: Option<ETypeId>) -> u64 {
  match etype_filter {
    Some(etype) => db.count_edges_by_type(etype) as u64,
    None => db.count_edges() as u64,
  }
}

// ============================================================================
// Schema Definitions
// ============================================================================

/// Property definition for nodes or edges
#[derive(Debug, Clone)]
pub struct PropDef {
  /// Property name
  pub name: String,
  /// Property type hint (for documentation/validation)
  pub prop_type: PropType,
  /// Whether this property is required
  pub required: bool,
  /// Default value (if any)
  pub default: Option<PropValue>,
}

/// Property type hints
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PropType {
  String,
  Int,
  Float,
  Bool,
  Any,
}

impl PropDef {
  pub fn string(name: &str) -> Self {
    Self {
      name: name.to_string(),
      prop_type: PropType::String,
      required: false,
      default: None,
    }
  }

  pub fn int(name: &str) -> Self {
    Self {
      name: name.to_string(),
      prop_type: PropType::Int,
      required: false,
      default: None,
    }
  }

  pub fn float(name: &str) -> Self {
    Self {
      name: name.to_string(),
      prop_type: PropType::Float,
      required: false,
      default: None,
    }
  }

  pub fn bool(name: &str) -> Self {
    Self {
      name: name.to_string(),
      prop_type: PropType::Bool,
      required: false,
      default: None,
    }
  }

  pub fn required(mut self) -> Self {
    self.required = true;
    self
  }

  pub fn default(mut self, value: PropValue) -> Self {
    self.default = Some(value);
    self
  }
}

/// Node type definition
#[derive(Debug, Clone)]
pub struct NodeDef {
  /// Node type name
  pub name: String,
  /// Property definitions
  pub props: HashMap<String, PropDef>,
  /// Key prefix for this node type (e.g., "user:")
  pub key_prefix: String,
  /// Internal label ID (set after registration)
  pub label_id: Option<LabelId>,
  /// Property key IDs (set after registration)
  pub prop_key_ids: HashMap<String, PropKeyId>,
}

impl NodeDef {
  pub fn new(name: &str, key_prefix: &str) -> Self {
    Self {
      name: name.to_string(),
      props: HashMap::new(),
      key_prefix: key_prefix.to_string(),
      label_id: None,
      prop_key_ids: HashMap::new(),
    }
  }

  pub fn prop(mut self, prop: PropDef) -> Self {
    self.props.insert(prop.name.clone(), prop);
    self
  }

  /// Generate a full key from a key suffix
  pub fn key(&self, suffix: &str) -> String {
    format!("{}{}", self.key_prefix, suffix)
  }
}

/// Edge type definition
#[derive(Debug, Clone)]
pub struct EdgeDef {
  /// Edge type name
  pub name: String,
  /// Property definitions
  pub props: HashMap<String, PropDef>,
  /// Internal edge type ID (set after registration)
  pub etype_id: Option<ETypeId>,
  /// Property key IDs (set after registration)
  pub prop_key_ids: HashMap<String, PropKeyId>,
}

impl EdgeDef {
  pub fn new(name: &str) -> Self {
    Self {
      name: name.to_string(),
      props: HashMap::new(),
      etype_id: None,
      prop_key_ids: HashMap::new(),
    }
  }

  pub fn prop(mut self, prop: PropDef) -> Self {
    self.props.insert(prop.name.clone(), prop);
    self
  }
}

// ============================================================================
// Node Reference
// ============================================================================

/// Reference to a node in the database
#[derive(Debug, Clone)]
pub struct NodeRef {
  /// Node ID
  id: NodeId,
  /// Full key (if available)
  key: Option<String>,
  /// Node type name
  node_type: Arc<str>,
}

impl NodeRef {
  pub fn new(id: NodeId, key: Option<String>, node_type: impl Into<Arc<str>>) -> Self {
    Self {
      id,
      key,
      node_type: node_type.into(),
    }
  }

  pub fn id(&self) -> NodeId {
    self.id
  }

  pub fn key(&self) -> Option<&str> {
    self.key.as_deref()
  }

  pub fn node_type(&self) -> &str {
    &self.node_type
  }

  pub fn into_parts(self) -> (NodeId, Option<String>, Arc<str>) {
    (self.id, self.key, self.node_type)
  }
}

// ============================================================================
// Kite Options
// ============================================================================

/// Options for opening a Kite database
#[derive(Debug, Clone)]
pub struct KiteOptions {
  /// Node type definitions
  pub nodes: Vec<NodeDef>,
  /// Edge type definitions
  pub edges: Vec<EdgeDef>,
  /// Open in read-only mode
  pub read_only: bool,
  /// Create database if it doesn't exist
  pub create_if_missing: bool,
  /// Synchronization mode for WAL writes (default: Full)
  pub sync_mode: SyncMode,
  /// Enable group commit (coalesce WAL flushes across commits)
  pub group_commit_enabled: bool,
  /// Group commit window in milliseconds
  pub group_commit_window_ms: u64,
  /// Enable MVCC (snapshot isolation + conflict detection)
  pub mvcc: bool,
  /// MVCC GC interval in ms
  pub mvcc_gc_interval_ms: Option<u64>,
  /// MVCC retention in ms
  pub mvcc_retention_ms: Option<u64>,
  /// MVCC max version chain depth
  pub mvcc_max_chain_depth: Option<usize>,
  /// WAL size in bytes (default: 4MB)
  pub wal_size: Option<usize>,
  /// WAL usage threshold (0.0-1.0) to trigger auto-checkpoint
  pub checkpoint_threshold: Option<f64>,
}

impl KiteOptions {
  pub fn new() -> Self {
    Self {
      nodes: Vec::new(),
      edges: Vec::new(),
      read_only: false,
      create_if_missing: true,
      sync_mode: SyncMode::Full,
      group_commit_enabled: false,
      group_commit_window_ms: 2,
      mvcc: false,
      mvcc_gc_interval_ms: None,
      mvcc_retention_ms: None,
      mvcc_max_chain_depth: None,
      wal_size: None,
      checkpoint_threshold: None,
    }
  }

  pub fn node(mut self, node: NodeDef) -> Self {
    self.nodes.push(node);
    self
  }

  pub fn edge(mut self, edge: EdgeDef) -> Self {
    self.edges.push(edge);
    self
  }

  pub fn read_only(mut self, value: bool) -> Self {
    self.read_only = value;
    self
  }

  pub fn sync_mode(mut self, mode: SyncMode) -> Self {
    self.sync_mode = mode;
    self
  }

  /// Enable or disable group commit (coalesce WAL flushes across commits)
  pub fn group_commit_enabled(mut self, value: bool) -> Self {
    self.group_commit_enabled = value;
    self
  }

  /// Set the group commit window in milliseconds
  pub fn group_commit_window_ms(mut self, value: u64) -> Self {
    self.group_commit_window_ms = value;
    self
  }

  /// Set sync mode to Normal (fsync on checkpoint only)
  pub fn sync_normal(mut self) -> Self {
    self.sync_mode = SyncMode::Normal;
    self
  }

  /// Set sync mode to Off (no fsync)
  pub fn sync_off(mut self) -> Self {
    self.sync_mode = SyncMode::Off;
    self
  }

  pub fn mvcc(mut self, value: bool) -> Self {
    self.mvcc = value;
    self
  }

  pub fn mvcc_gc_interval_ms(mut self, value: u64) -> Self {
    self.mvcc_gc_interval_ms = Some(value);
    self
  }

  pub fn mvcc_retention_ms(mut self, value: u64) -> Self {
    self.mvcc_retention_ms = Some(value);
    self
  }

  pub fn mvcc_max_chain_depth(mut self, value: usize) -> Self {
    self.mvcc_max_chain_depth = Some(value);
    self
  }

  /// Set WAL size in bytes
  pub fn wal_size(mut self, value: usize) -> Self {
    self.wal_size = Some(value);
    self
  }

  /// Set WAL size in megabytes
  pub fn wal_size_mb(mut self, value: usize) -> Self {
    self.wal_size = Some(value.saturating_mul(1024 * 1024));
    self
  }

  /// Set checkpoint threshold (0.0-1.0)
  pub fn checkpoint_threshold(mut self, value: f64) -> Self {
    self.checkpoint_threshold = Some(value.clamp(0.0, 1.0));
    self
  }
}

impl Default for KiteOptions {
  fn default() -> Self {
    Self::new()
  }
}

/// Convenience helper to open a KiteDB instance.
pub fn kite<P: AsRef<Path>>(path: P, options: KiteOptions) -> Result<Kite> {
  Kite::open(path, options)
}

// ============================================================================
// Kite Database
// ============================================================================

/// High-level graph database API
pub struct Kite {
  /// Underlying database
  db: SingleFileDB,
  /// Node type definitions by name
  nodes: HashMap<String, NodeDef>,
  /// Edge type definitions by name
  edges: HashMap<String, EdgeDef>,
  /// Key prefix to node def mapping for fast lookups
  key_prefix_to_node: HashMap<String, String>,
}

impl Kite {
  /// Open or create a Kite database
  pub fn open<P: AsRef<Path>>(path: P, options: KiteOptions) -> Result<Self> {
    let path = path.as_ref();
    if path.exists() && path.is_dir() {
      return Err(KiteError::InvalidPath(
        "Directory-format databases are no longer supported; use a .kitedb file path".to_string(),
      ));
    }

    let mut db_path = PathBuf::from(path);
    if db_path.extension().is_some() {
      if !is_single_file_path(&db_path) {
        let ext = db_path
          .extension()
          .map(|value| value.to_string_lossy())
          .unwrap_or_else(|| "".into());
        return Err(KiteError::InvalidPath(format!(
          "Invalid database extension '.{ext}'. Single-file databases must use {} (or pass a path without an extension).",
          single_file_extension()
        )));
      }
    } else {
      db_path = PathBuf::from(format!("{}{}", path.display(), single_file_extension()));
    }

    let mut db_options = SingleFileOpenOptions::new()
      .read_only(options.read_only)
      .create_if_missing(options.create_if_missing)
      .sync_mode(options.sync_mode)
      .group_commit_enabled(options.group_commit_enabled)
      .group_commit_window_ms(options.group_commit_window_ms)
      .mvcc(options.mvcc);
    if let Some(v) = options.mvcc_gc_interval_ms {
      db_options = db_options.mvcc_gc_interval_ms(v);
    }
    if let Some(v) = options.mvcc_retention_ms {
      db_options = db_options.mvcc_retention_ms(v);
    }
    if let Some(v) = options.mvcc_max_chain_depth {
      db_options = db_options.mvcc_max_chain_depth(v);
    }
    if let Some(v) = options.wal_size {
      db_options = db_options.wal_size(v);
    }
    if let Some(v) = options.checkpoint_threshold {
      db_options = db_options.checkpoint_threshold(v);
    }
    let db = open_single_file(&db_path, db_options)?;

    // Initialize schema in a transaction
    let mut nodes: HashMap<String, NodeDef> = HashMap::new();
    let mut edges: HashMap<String, EdgeDef> = HashMap::new();
    let mut key_prefix_to_node: HashMap<String, String> = HashMap::new();

    // Process node definitions
    for mut node_def in options.nodes {
      // Define label
      let label_id = db.label_id_or_create(&node_def.name);
      node_def.label_id = Some(label_id);

      // Define property keys
      for prop_name in node_def.props.keys() {
        let prop_key_id = db.propkey_id_or_create(prop_name);
        node_def.prop_key_ids.insert(prop_name.clone(), prop_key_id);
      }

      key_prefix_to_node.insert(node_def.key_prefix.clone(), node_def.name.clone());
      nodes.insert(node_def.name.clone(), node_def);
    }

    // Process edge definitions
    for mut edge_def in options.edges {
      // Define edge type
      let etype_id = db.etype_id_or_create(&edge_def.name);
      edge_def.etype_id = Some(etype_id);

      // Define property keys
      for prop_name in edge_def.props.keys() {
        let prop_key_id = db.propkey_id_or_create(prop_name);
        edge_def.prop_key_ids.insert(prop_name.clone(), prop_key_id);
      }

      edges.insert(edge_def.name.clone(), edge_def);
    }

    Ok(Self {
      db,
      nodes,
      edges,
      key_prefix_to_node,
    })
  }

  // ========================================================================
  // Node Operations
  // ========================================================================

  /// Create a new node
  pub fn create_node(
    &mut self,
    node_type: &str,
    key_suffix: &str,
    props: HashMap<String, PropValue>,
  ) -> Result<NodeRef> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .clone();

    let full_key = node_def.key(key_suffix);

    // Begin transaction
    let mut handle = begin_tx(&self.db)?;

    // Create the node with key
    let node_opts = NodeOpts {
      key: Some(full_key.clone()),
      labels: node_def.label_id.map(|id| vec![id]),
      props: None,
    };
    let node_id = create_node(&mut handle, node_opts)?;

    // Set properties
    for (prop_name, value) in props {
      if let Some(&prop_key_id) = node_def.prop_key_ids.get(&prop_name) {
        set_node_prop(&mut handle, node_id, prop_key_id, value)?;
      }
    }

    // Commit
    commit(&mut handle)?;

    Ok(NodeRef::new(node_id, Some(full_key), node_type))
  }

  /// Insert a node using fluent builder API
  ///
  /// This method provides a more ergonomic way to create nodes with properties
  /// using the builder pattern. Use `.values()` to specify the node data,
  /// then either `.execute()` to insert without returning, or `.returning()`
  /// to get the created node reference.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::PropValue;
  /// # use std::collections::HashMap;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// # let props: HashMap<String, PropValue> = HashMap::new();
  /// // Insert and get the node reference
  /// let user = kite.insert("User")?
  ///     .values("alice", props)?
  ///     .returning()?;
  ///
  /// // Insert without returning (slightly faster)
  /// kite.insert("User")?
  ///     .values("bob", HashMap::new())?
  ///     .execute()?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn insert(&mut self, node_type: &str) -> Result<KiteInsertBuilder<'_>> {
    let key_prefix = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .key_prefix
      .clone();

    Ok(KiteInsertBuilder {
      ray: self,
      node_type: node_type.to_string(),
      key_prefix,
    })
  }

  /// Upsert a node using fluent builder API
  ///
  /// Creates the node if it doesn't exist, otherwise updates properties
  /// on the existing node.
  pub fn upsert(&mut self, node_type: &str) -> Result<KiteUpsertBuilder<'_>> {
    let key_prefix = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .key_prefix
      .clone();

    Ok(KiteUpsertBuilder {
      ray: self,
      node_type: node_type.to_string(),
      key_prefix,
    })
  }

  /// Get a node by key (direct read, no transaction overhead)
  pub fn get(&self, node_type: &str, key_suffix: &str) -> Result<Option<NodeRef>> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?;

    let full_key = node_def.key(key_suffix);

    // Direct read without transaction
    let node_id = node_by_key_db(&self.db, &full_key);

    match node_id {
      Some(id) => Ok(Some(NodeRef::new(id, Some(full_key), node_type))),
      None => Ok(None),
    }
  }

  /// Get a node by ID (direct read, no transaction overhead)
  pub fn node_by_id(&self, node_id: NodeId) -> Result<Option<NodeRef>> {
    // Direct read without transaction
    let exists = node_exists_db(&self.db, node_id);

    if exists {
      // Look up the node's key from snapshot/delta
      let key = self.db.node_key(node_id);

      // Try to determine node type from key prefix
      let node_type = if let Some(ref k) = key {
        // Find matching node def by key prefix
        self
          .nodes
          .values()
          .find(|def| k.starts_with(&def.key_prefix))
          .map(|def| def.name.as_str())
          .unwrap_or("unknown")
      } else {
        "unknown"
      };

      Ok(Some(NodeRef::new(node_id, key, node_type)))
    } else {
      Ok(None)
    }
  }

  /// Check if a node exists (direct read, no transaction overhead)
  pub fn exists(&self, node_id: NodeId) -> bool {
    // Direct read without transaction
    node_exists_db(&self.db, node_id)
  }

  /// Delete a node
  pub fn delete_node(&mut self, node_id: NodeId) -> Result<bool> {
    let mut handle = begin_tx(&self.db)?;
    let deleted = delete_node(&mut handle, node_id)?;
    commit(&mut handle)?;
    Ok(deleted)
  }

  /// Get a node property (direct read, no transaction overhead)
  pub fn prop(&self, node_id: NodeId, prop_name: &str) -> Option<PropValue> {
    let prop_key_id = self.db.propkey_id(prop_name)?;
    // Direct read without transaction
    node_prop_db(&self.db, node_id, prop_key_id)
  }

  /// Set a node property
  pub fn set_prop(&mut self, node_id: NodeId, prop_name: &str, value: PropValue) -> Result<()> {
    let prop_key_id = self.db.propkey_id_or_create(prop_name);

    let mut handle = begin_tx(&self.db)?;
    set_node_prop(&mut handle, node_id, prop_key_id, value)?;
    commit(&mut handle)?;
    Ok(())
  }

  /// Set multiple node properties in a single transaction
  pub fn set_props<I, K>(&mut self, node_id: NodeId, props: I) -> Result<()>
  where
    I: IntoIterator<Item = (K, PropValue)>,
    K: AsRef<str>,
  {
    let mut iter = props.into_iter();
    let Some((first_name, first_value)) = iter.next() else {
      return Ok(());
    };

    let mut handle = begin_tx(&self.db)?;

    let first_key_id = self.db.propkey_id_or_create(first_name.as_ref());
    set_node_prop(&mut handle, node_id, first_key_id, first_value)?;

    for (prop_name, value) in iter {
      let prop_key_id = self.db.propkey_id_or_create(prop_name.as_ref());
      set_node_prop(&mut handle, node_id, prop_key_id, value)?;
    }

    commit(&mut handle)?;
    Ok(())
  }

  /// Update a node by reference using fluent builder API
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::PropValue;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// let alice = match kite.get("User", "alice")? {
  ///   Some(node) => node,
  ///   None => return Ok(()),
  /// };
  /// kite.update(&alice)?
  ///     .set("name", PropValue::String("Alice Updated".into()))
  ///     .set("age", PropValue::I64(31))
  ///     .execute()?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn update(&mut self, node_ref: &NodeRef) -> Result<KiteUpdateNodeBuilder<'_>> {
    // Verify node exists
    let exists = {
      let mut handle = begin_tx(&self.db)?;
      let exists = node_exists(&handle, node_ref.id());
      commit(&mut handle)?;
      exists
    };

    if !exists {
      return Err(KiteError::NodeNotFound(node_ref.id()));
    }

    Ok(KiteUpdateNodeBuilder {
      ray: self,
      node_id: node_ref.id(),
      updates: HashMap::new(),
    })
  }

  /// Update a node by ID using fluent builder API
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::{NodeId, PropValue};
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// # let node_id: NodeId = 1;
  /// kite.update_by_id(node_id)?
  ///     .set("name", PropValue::String("Updated".into()))
  ///     .execute()?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn update_by_id(&mut self, node_id: NodeId) -> Result<KiteUpdateNodeBuilder<'_>> {
    // Verify node exists
    let exists = {
      let mut handle = begin_tx(&self.db)?;
      let exists = node_exists(&handle, node_id);
      commit(&mut handle)?;
      exists
    };

    if !exists {
      return Err(KiteError::NodeNotFound(node_id));
    }

    Ok(KiteUpdateNodeBuilder {
      ray: self,
      node_id,
      updates: HashMap::new(),
    })
  }

  /// Upsert a node by ID using fluent builder API
  ///
  /// Creates the node if missing, otherwise updates properties.
  pub fn upsert_by_id(
    &mut self,
    node_type: &str,
    node_id: NodeId,
  ) -> Result<KiteUpsertByIdBuilder<'_>> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .clone();

    Ok(KiteUpsertByIdBuilder {
      ray: self,
      node_id,
      node_def,
      updates: HashMap::new(),
    })
  }

  /// Update a node by key using fluent builder API
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::PropValue;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// kite.update_by_key("User", "alice")?
  ///     .set("name", PropValue::String("Alice Updated".into()))
  ///     .execute()?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn update_by_key(
    &mut self,
    node_type: &str,
    key_suffix: &str,
  ) -> Result<KiteUpdateNodeBuilder<'_>> {
    let full_key = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .key(key_suffix);

    let node_id = {
      let mut handle = begin_tx(&self.db)?;
      let node_id =
        node_by_key(&handle, &full_key).ok_or_else(|| KiteError::KeyNotFound(full_key.clone()))?;
      commit(&mut handle)?;
      node_id
    };

    Ok(KiteUpdateNodeBuilder {
      ray: self,
      node_id,
      updates: HashMap::new(),
    })
  }

  // ========================================================================
  // Edge Operations
  // ========================================================================

  /// Create an edge between two nodes
  pub fn link(&mut self, src: NodeId, edge_type: &str, dst: NodeId) -> Result<()> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let mut handle = begin_tx(&self.db)?;
    add_edge(&mut handle, src, etype_id, dst)?;
    commit(&mut handle)?;
    Ok(())
  }

  /// Create an edge between two nodes with properties
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::{NodeRef, Kite};
  /// # use kitedb::types::PropValue;
  /// # use std::collections::HashMap;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// # let alice: NodeRef = unimplemented!();
  /// # let bob: NodeRef = unimplemented!();
  /// let mut props = HashMap::new();
  /// props.insert("weight".to_string(), PropValue::F64(0.5));
  /// props.insert("since".to_string(), PropValue::String("2024".into()));
  /// kite.link_with_props(alice.id(), "FOLLOWS", bob.id(), props)?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn link_with_props(
    &mut self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
    props: HashMap<String, PropValue>,
  ) -> Result<()> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?
      .clone();

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let mut handle = begin_tx(&self.db)?;
    if props.is_empty() {
      add_edge(&mut handle, src, etype_id, dst)?;
    } else {
      let mut prop_pairs = Vec::with_capacity(props.len());
      for (prop_name, value) in props {
        let prop_key_id = if let Some(&id) = edge_def.prop_key_ids.get(&prop_name) {
          id
        } else {
          handle.db.propkey_id_or_create(&prop_name)
        };
        prop_pairs.push((prop_key_id, value));
      }
      handle
        .db
        .add_edge_with_props(src, etype_id, dst, prop_pairs)?;
    }

    commit(&mut handle)?;
    Ok(())
  }

  /// Remove an edge between two nodes
  pub fn unlink(&mut self, src: NodeId, edge_type: &str, dst: NodeId) -> Result<bool> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let mut handle = begin_tx(&self.db)?;
    let deleted = delete_edge(&mut handle, src, etype_id, dst)?;
    commit(&mut handle)?;
    Ok(deleted)
  }

  /// Check if an edge exists (direct read, no transaction overhead)
  pub fn has_edge(&self, src: NodeId, edge_type: &str, dst: NodeId) -> Result<bool> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    // Direct read without transaction
    Ok(edge_exists_db(&self.db, src, etype_id, dst))
  }

  /// Get outgoing neighbors of a node (direct read, no transaction overhead)
  pub fn neighbors_out(&self, node_id: NodeId, edge_type: Option<&str>) -> Result<Vec<NodeId>> {
    let etype_id = match edge_type {
      Some(name) => {
        let edge_def = self
          .edges
          .get(name)
          .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {name}").into()))?;
        edge_def.etype_id
      }
      None => None,
    };

    // Direct read without transaction
    Ok(neighbors_out_db(&self.db, node_id, etype_id))
  }

  /// Get incoming neighbors of a node (direct read, no transaction overhead)
  pub fn neighbors_in(&self, node_id: NodeId, edge_type: Option<&str>) -> Result<Vec<NodeId>> {
    let etype_id = match edge_type {
      Some(name) => {
        let edge_def = self
          .edges
          .get(name)
          .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {name}").into()))?;
        edge_def.etype_id
      }
      None => None,
    };

    // Direct read without transaction
    let neighbors = neighbors_in_db(&self.db, node_id, etype_id);
    Ok(neighbors)
  }

  // ========================================================================
  // Edge Property Operations
  // ========================================================================

  /// Get an edge property (direct read, no transaction overhead)
  ///
  /// Returns None if the edge doesn't exist or the property is not set.
  pub fn edge_prop(
    &self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
    prop_name: &str,
  ) -> Result<Option<PropValue>> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let prop_key_id = match self.db.propkey_id(prop_name) {
      Some(id) => id,
      None => return Ok(None), // Unknown property = not set
    };

    // Direct read without transaction
    Ok(edge_prop_db(&self.db, src, etype_id, dst, prop_key_id))
  }

  /// Get all properties for an edge (direct read, no transaction overhead)
  ///
  /// Returns None if the edge doesn't exist.
  pub fn edge_props(
    &self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
  ) -> Result<Option<HashMap<String, PropValue>>> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    // Direct read without transaction
    let props = edge_props_db(&self.db, src, etype_id, dst);

    // Convert PropKeyId -> String in the result
    match props {
      Some(props_by_id) => {
        let mut result = HashMap::new();
        for (key_id, value) in props_by_id {
          if let Some(name) = self.db.propkey_name(key_id) {
            result.insert(name, value);
          }
        }
        Ok(Some(result))
      }
      None => Ok(None),
    }
  }

  /// Set an edge property
  pub fn set_edge_prop(
    &mut self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
    prop_name: &str,
    value: PropValue,
  ) -> Result<()> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let prop_key_id = self.db.propkey_id_or_create(prop_name);

    let mut handle = begin_tx(&self.db)?;
    set_edge_prop(&mut handle, src, etype_id, dst, prop_key_id, value)?;
    commit(&mut handle)?;
    Ok(())
  }

  /// Set multiple edge properties
  pub fn set_edge_props(
    &mut self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
    props: HashMap<String, PropValue>,
  ) -> Result<()> {
    if props.is_empty() {
      return Ok(());
    }

    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let mut prop_pairs = Vec::with_capacity(props.len());
    for (prop_name, value) in props {
      let prop_key_id = if let Some(&id) = edge_def.prop_key_ids.get(&prop_name) {
        id
      } else {
        self.db.propkey_id_or_create(&prop_name)
      };
      prop_pairs.push((prop_key_id, value));
    }

    let mut handle = begin_tx(&self.db)?;
    handle.db.set_edge_props(src, etype_id, dst, prop_pairs)?;
    commit(&mut handle)?;
    Ok(())
  }

  /// Delete an edge property
  pub fn del_edge_prop(
    &mut self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
    prop_name: &str,
  ) -> Result<()> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    let prop_key_id = self
      .db
      .propkey_id(prop_name)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown property: {prop_name}").into()))?;

    let mut handle = begin_tx(&self.db)?;
    del_edge_prop(&mut handle, src, etype_id, dst, prop_key_id)?;
    commit(&mut handle)?;
    Ok(())
  }

  /// Update edge properties using fluent builder API
  ///
  /// Returns an `UpdateEdgeBuilder` that allows setting multiple properties
  /// in a single transaction.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::{NodeId, PropValue};
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// # let alice_id: NodeId = 1;
  /// # let bob_id: NodeId = 2;
  /// kite.update_edge(alice_id, "FOLLOWS", bob_id)?
  ///    .set("weight", PropValue::F64(0.9))
  ///    .set("since", PropValue::String("2024".to_string()))
  ///    .execute()?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn update_edge(
    &mut self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
  ) -> Result<KiteUpdateEdgeBuilder<'_>> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    Ok(KiteUpdateEdgeBuilder {
      ray: self,
      src,
      etype_id,
      dst,
      updates: HashMap::new(),
    })
  }

  /// Upsert edge properties using fluent builder API
  ///
  /// Creates the edge if missing, otherwise updates properties.
  pub fn upsert_edge(
    &mut self,
    src: NodeId,
    edge_type: &str,
    dst: NodeId,
  ) -> Result<KiteUpsertEdgeBuilder<'_>> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;
    let etype_id = edge_def.etype_id.ok_or_else(|| {
      KiteError::InvalidSchema(format!("Edge type not initialized: {edge_type}").into())
    })?;

    Ok(KiteUpsertEdgeBuilder {
      ray: self,
      src,
      etype_id,
      dst,
      updates: HashMap::new(),
    })
  }

  // ========================================================================
  // Listing and Counting
  // ========================================================================

  /// Count all nodes in the database
  ///
  /// This is an O(1) operation when possible, using cached counts.
  pub fn count_nodes(&self) -> u64 {
    count_nodes(&self.db)
  }

  /// Count nodes of a specific type
  ///
  /// This requires iteration to filter by key prefix.
  pub fn count_nodes_by_type(&self, node_type: &str) -> Result<u64> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?;

    let prefix = &node_def.key_prefix;
    let mut count = 0u64;

    for node_id in list_nodes(&self.db) {
      if let Some(key) = self.node_key_internal(node_id) {
        if key.starts_with(prefix) {
          count += 1;
        }
      }
    }

    Ok(count)
  }

  /// Count all edges
  pub fn count_edges(&self) -> u64 {
    count_edges(&self.db, None)
  }

  /// Count edges of a specific type
  pub fn count_edges_by_type(&self, edge_type: &str) -> Result<u64> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    Ok(count_edges(&self.db, Some(etype_id)))
  }

  #[cfg(feature = "bench-profile")]
  pub fn take_profile_snapshot(&self) -> (u64, u64) {
    self.db.take_profile_snapshot()
  }

  /// List all node IDs
  pub fn list_nodes(&self) -> Vec<NodeId> {
    list_nodes(&self.db)
  }

  /// Iterate over all nodes of a specific type
  ///
  /// Returns an iterator that yields `NodeRef` for each matching node.
  /// Filters nodes by matching their key prefix.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// for node_ref in kite.all("User")? {
  ///     println!("User: {:?}", node_ref.id());
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn all(&self, node_type: &str) -> Result<impl Iterator<Item = NodeRef> + '_> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .clone();

    let prefix = node_def.key_prefix.clone();
    let node_type_arc: Arc<str> = node_type.to_string().into();

    Ok(list_nodes(&self.db).into_iter().filter_map(move |node_id| {
      let key = self.node_key_internal(node_id)?;
      if key.starts_with(&prefix) {
        Some(NodeRef::new(node_id, Some(key), Arc::clone(&node_type_arc)))
      } else {
        None
      }
    }))
  }

  /// List all edges in the database
  pub fn list_all_edges(&self) -> Vec<FullEdge> {
    list_edges(&self.db, ListEdgesOptions::default())
  }

  /// Iterate over all edges, optionally filtered by type
  ///
  /// Returns an iterator that yields edge information.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// for edge in kite.all_edges(Some("FOLLOWS"))? {
  ///     println!("{} -> {}", edge.src, edge.dst);
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn all_edges(&self, edge_type: Option<&str>) -> Result<impl Iterator<Item = FullEdge> + '_> {
    let etype_id = match edge_type {
      Some(name) => {
        let edge_def = self
          .edges
          .get(name)
          .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {name}").into()))?;
        edge_def.etype_id
      }
      None => None,
    };

    let options = ListEdgesOptions { etype: etype_id };
    Ok(list_edges(&self.db, options).into_iter())
  }

  /// Get a lightweight node reference without loading properties
  ///
  /// This is faster than `get()` when you only need the node reference
  /// for traversals or edge operations.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// let user_ref = kite.node_ref("User", "alice")?;
  /// if let Some(node) = user_ref {
  ///     // Can now use node.id() for edges, traversals, etc.
  /// }
  /// # Ok(())
  /// # }
  /// ```
  /// Get a lightweight node reference by key (direct read, no transaction overhead)
  ///
  /// This is faster than `get()` as it only returns a reference without loading properties.
  /// Use this when you only need the node ID for traversals or edge operations.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// // Fast: only gets reference (~85ns)
  /// if let Some(node) = kite.node_ref("User", "alice")? {
  ///     // Can now use node.id() for edges, traversals, etc.
  ///     let friends = kite.from(node.id()).out(Some("FOLLOWS"))?.to_vec();
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn node_ref(&self, node_type: &str, key_suffix: &str) -> Result<Option<NodeRef>> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?;

    let full_key = node_def.key(key_suffix);

    // Direct read without transaction
    let node_id = node_by_key_db(&self.db, &full_key);

    match node_id {
      Some(id) => Ok(Some(NodeRef::new(id, Some(full_key), node_type))),
      None => Ok(None),
    }
  }

  /// Helper to get node key from database
  fn node_key_internal(&self, node_id: NodeId) -> Option<String> {
    self.db.node_key(node_id)
  }

  // ========================================================================
  // Schema Access
  // ========================================================================

  /// Get a node definition by name
  pub fn node_def(&self, name: &str) -> Option<&NodeDef> {
    self.nodes.get(name)
  }

  /// Get an edge definition by name
  pub fn edge_def(&self, name: &str) -> Option<&EdgeDef> {
    self.edges.get(name)
  }

  /// Get all node type names
  pub fn node_types(&self) -> Vec<&str> {
    self.nodes.keys().map(|s| s.as_str()).collect()
  }

  /// Get all edge type names
  pub fn edge_types(&self) -> Vec<&str> {
    self.edges.keys().map(|s| s.as_str()).collect()
  }

  // ========================================================================
  // Traversal
  // ========================================================================

  /// Start a traversal from a node
  ///
  /// Returns a traversal builder that can be used to chain traversal steps.
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use kitedb::api::kite::{NodeRef, Kite};
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// # let alice: NodeRef = unimplemented!();
  /// let friends = kite
  ///     .from(alice.id())
  ///     .out(Some("FOLLOWS"))?
  ///     .out(Some("FOLLOWS"))?
  ///     .to_vec();
  /// # Ok(())
  /// # }
  /// ```
  pub fn from(&self, node_id: NodeId) -> KiteTraversalBuilder<'_> {
    KiteTraversalBuilder::new(self, vec![node_id])
  }

  /// Start a traversal from multiple nodes
  pub fn from_nodes(&self, node_ids: Vec<NodeId>) -> KiteTraversalBuilder<'_> {
    KiteTraversalBuilder::new(self, node_ids)
  }

  // ========================================================================
  // Pathfinding
  // ========================================================================

  /// Find the shortest path between two nodes
  ///
  /// Returns a path finding builder that can be configured with edge types,
  /// direction, and maximum depth.
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use kitedb::api::kite::{NodeRef, Kite};
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// # let alice: NodeRef = unimplemented!();
  /// # let bob: NodeRef = unimplemented!();
  /// let path = kite
  ///     .shortest_path(alice.id(), bob.id())
  ///     .via("FOLLOWS")?
  ///     .max_depth(5)
  ///     .find();
  ///
  /// if path.found {
  ///     println!("Path: {:?}", path.path);
  ///     println!("Total weight: {}", path.total_weight);
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn shortest_path(&self, source: NodeId, target: NodeId) -> KitePathBuilder<'_> {
    KitePathBuilder::new(self, source, target)
  }

  /// Find shortest paths to any of the target nodes
  pub fn shortest_path_to_any(&self, source: NodeId, targets: Vec<NodeId>) -> KitePathBuilder<'_> {
    KitePathBuilder::new_multi(self, source, targets)
  }

  /// Check if a path exists between two nodes
  ///
  /// This is more efficient than `shortest_path()` when you only need to
  /// know if a path exists, not the path itself.
  pub fn has_path(
    &mut self,
    source: NodeId,
    target: NodeId,
    edge_type: Option<&str>,
  ) -> Result<bool> {
    let path = self.shortest_path(source, target);
    let path = if let Some(etype) = edge_type {
      path.via(etype)?
    } else {
      path
    };
    Ok(path.find().found)
  }

  /// Get all nodes reachable from a source within a certain depth
  ///
  /// # Example
  ///
  /// ```rust,no_run
  /// # use kitedb::api::kite::{NodeRef, Kite};
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// # let alice: NodeRef = unimplemented!();
  /// let reachable = kite.reachable_from(alice.id(), 3, Some("FOLLOWS"))?;
  /// println!("Alice can reach {} nodes in 3 hops", reachable.len());
  /// # Ok(())
  /// # }
  /// ```
  pub fn reachable_from(
    &self,
    source: NodeId,
    max_depth: usize,
    edge_type: Option<&str>,
  ) -> Result<Vec<NodeId>> {
    let etype = match edge_type {
      Some(name) => {
        let edge_def = self
          .edges
          .get(name)
          .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {name}").into()))?;
        edge_def.etype_id
      }
      None => None,
    };

    use super::traversal::{TraversalBuilder, TraversalDirection, TraverseOptions};

    let options = TraverseOptions::new(TraversalDirection::Out, max_depth);

    let results = TraversalBuilder::from_node(source)
      .traverse(etype, options)
      .collect_node_ids(|node_id, dir, etype_filter| self.neighbors(node_id, dir, etype_filter));

    Ok(results)
  }

  // Internal helper to get neighbors for traversal/pathfinding (read-only, no transaction)
  fn neighbors(
    &self,
    node_id: NodeId,
    direction: super::traversal::TraversalDirection,
    etype: Option<ETypeId>,
  ) -> Vec<Edge> {
    use super::traversal::TraversalDirection;

    let mut edges = Vec::new();

    match direction {
      TraversalDirection::Out => {
        for (edge_etype, dst) in self.db.out_edges(node_id) {
          if etype.is_some() && etype != Some(edge_etype) {
            continue;
          }
          edges.push(Edge {
            src: node_id,
            etype: edge_etype,
            dst,
          });
        }
      }
      TraversalDirection::In => {
        for (edge_etype, src) in self.db.in_edges(node_id) {
          if etype.is_some() && etype != Some(edge_etype) {
            continue;
          }
          edges.push(Edge {
            src,
            etype: edge_etype,
            dst: node_id,
          });
        }
      }
      TraversalDirection::Both => {
        edges.extend(self.neighbors(node_id, TraversalDirection::Out, etype));
        edges.extend(self.neighbors(node_id, TraversalDirection::In, etype));
      }
    }

    edges
  }

  // ========================================================================
  // Database Maintenance
  // ========================================================================

  /// Optimize (compact) the database
  ///
  /// This merges the write-ahead log (WAL) into the snapshot, reducing
  /// file size and improving read performance. This is equivalent to
  /// "VACUUM" in SQLite.
  /// Call this periodically to reclaim space from deleted nodes/edges
  /// and improve read performance.
  pub fn optimize(&mut self) -> Result<()> {
    self.db.optimize_single_file(None)
  }

  /// Get database statistics
  pub fn stats(&self) -> DbStats {
    self.db.stats()
  }

  /// Get a human-readable description of the database
  ///
  /// Useful for debugging and monitoring. Returns information about:
  /// - Database path and format
  /// - Schema (node types and edge types)
  /// - Current statistics
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # fn main() {
  /// # let kite: Kite = unimplemented!();
  /// println!("{}", kite.describe());
  /// // Output:
  /// // KiteDB at /path/to/db.kitedb (single-file format)
  /// // Schema:
  /// //   Node types: User, Post, Comment
  /// //   Edge types: FOLLOWS, LIKES, WROTE
  /// // Statistics:
  /// //   Nodes: 1,234 (snapshot: 1,200, delta: +34)
  /// //   Edges: 5,678 (snapshot: 5,600, delta: +78)
  /// # }
  /// ```
  pub fn describe(&self) -> String {
    let stats = self.stats();
    let path = self.db.path.display();
    let format = "single-file";

    let node_types: Vec<&str> = self.nodes.keys().map(|s| s.as_str()).collect();
    let edge_types: Vec<&str> = self.edges.keys().map(|s| s.as_str()).collect();

    let delta_nodes = stats.delta_nodes_created as i64 - stats.delta_nodes_deleted as i64;
    let delta_edges = stats.delta_edges_added as i64 - stats.delta_edges_deleted as i64;

    format!(
      "KiteDB at {} ({} format)\n\
       Schema:\n  \
         Node types: {}\n  \
         Edge types: {}\n\
       Statistics:\n  \
         Nodes: {} (snapshot: {}, delta: {:+})\n  \
         Edges: {} (snapshot: {}, delta: {:+})\n  \
         Recommend compact: {}",
      path,
      format,
      if node_types.is_empty() {
        "(none)".to_string()
      } else {
        node_types.join(", ")
      },
      if edge_types.is_empty() {
        "(none)".to_string()
      } else {
        edge_types.join(", ")
      },
      stats.snapshot_nodes,
      stats
        .snapshot_nodes
        .saturating_sub(stats.delta_nodes_created as u64),
      delta_nodes,
      stats.snapshot_edges,
      stats
        .snapshot_edges
        .saturating_sub(stats.delta_edges_added as u64),
      delta_edges,
      if stats.recommend_compact { "yes" } else { "no" }
    )
  }

  /// Check database integrity
  ///
  /// Performs validation checks on the database structure:
  /// - Verifies edge reciprocity (for each outgoing edge, a matching incoming edge exists)
  /// - Checks that all edges reference existing nodes
  /// - Validates node key mappings
  ///
  /// Returns a `CheckResult` with `valid=true` if no errors found, or detailed
  /// error/warning messages otherwise.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// let result = kite.check()?;
  /// if !result.valid {
  ///     for error in &result.errors {
  ///         eprintln!("Error: {}", error);
  ///     }
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn check(&self) -> Result<CheckResult> {
    let mut result = self.db.check();

    // Schema consistency - verify all registered edge types have valid IDs
    for (edge_name, edge_def) in &self.edges {
      if edge_def.etype_id.is_none() {
        result
          .warnings
          .push(format!("Edge type '{edge_name}' has no assigned etype_id"));
      }
    }

    Ok(result)
  }

  // ========================================================================
  // Database Access
  // ========================================================================

  /// Get a reference to the underlying SingleFileDB
  pub fn raw(&self) -> &SingleFileDB {
    &self.db
  }

  /// Get a mutable reference to the underlying SingleFileDB
  pub fn raw_mut(&mut self) -> &mut SingleFileDB {
    &mut self.db
  }

  /// Close the database
  pub fn close(self) -> Result<()> {
    close_single_file(self.db)
  }
}

// ============================================================================
// Traversal Builder for Kite
// ============================================================================

use super::traversal::{TraversalBuilder, TraversalDirection, TraversalResult, TraverseOptions};

/// Traversal builder bound to a Kite database
///
/// Provides ergonomic traversal operations using edge type names.
pub struct KiteTraversalBuilder<'a> {
  ray: &'a Kite,
  builder: TraversalBuilder,
}

impl<'a> KiteTraversalBuilder<'a> {
  fn new(ray: &'a Kite, start_nodes: Vec<NodeId>) -> Self {
    Self {
      ray,
      builder: TraversalBuilder::new(start_nodes),
    }
  }

  /// Traverse outgoing edges
  ///
  /// @param edge_type - Edge type name (or None for all types)
  pub fn out(mut self, edge_type: Option<&str>) -> Result<Self> {
    let etype = self.resolve_etype(edge_type)?;
    self.builder = self.builder.out(etype);
    Ok(self)
  }

  /// Traverse incoming edges
  pub fn r#in(mut self, edge_type: Option<&str>) -> Result<Self> {
    let etype = self.resolve_etype(edge_type)?;
    self.builder = self.builder.r#in(etype);
    Ok(self)
  }

  /// Traverse edges in both directions
  pub fn both(mut self, edge_type: Option<&str>) -> Result<Self> {
    let etype = self.resolve_etype(edge_type)?;
    self.builder = self.builder.both(etype);
    Ok(self)
  }

  /// Variable-depth traversal
  pub fn traverse(mut self, edge_type: Option<&str>, options: TraverseOptions) -> Result<Self> {
    let etype = self.resolve_etype(edge_type)?;
    self.builder = self.builder.traverse(etype, options);
    Ok(self)
  }

  /// Limit the number of results
  pub fn take(mut self, limit: usize) -> Self {
    self.builder = self.builder.take(limit);
    self
  }

  /// Select specific properties to load (optimization)
  ///
  /// Only the specified properties will be loaded when collecting results,
  /// reducing overhead. This is useful when you only need a few properties
  /// from nodes that have many properties.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::NodeId;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// # let user_id: NodeId = 1;
  /// let friends = kite.from(user_id)
  ///     .out(Some("FOLLOWS"))?
  ///     .select(&["name", "avatar"]) // Only load name and avatar
  ///     .to_vec();
  /// # Ok(())
  /// # }
  /// ```
  pub fn select(mut self, props: &[&str]) -> Self {
    self.builder = self.builder.select_props(props);
    self
  }

  /// Execute and collect node IDs
  pub fn to_vec(self) -> Vec<NodeId> {
    self
      .builder
      .collect_node_ids(|node_id, dir, etype| self.ray.neighbors(node_id, dir, etype))
  }

  /// Execute and get first result
  pub fn first(self) -> Option<TraversalResult> {
    self
      .builder
      .first(|node_id, dir, etype| self.ray.neighbors(node_id, dir, etype))
  }

  /// Execute and get first node ID
  pub fn first_node(self) -> Option<NodeId> {
    self
      .builder
      .first_node(|node_id, dir, etype| self.ray.neighbors(node_id, dir, etype))
  }

  /// Execute and count results
  pub fn count(self) -> usize {
    self
      .builder
      .count(|node_id, dir, etype| self.ray.neighbors(node_id, dir, etype))
  }

  /// Execute and return iterator over traversal results
  pub fn execute(self) -> impl Iterator<Item = TraversalResult> + 'a {
    let ray = self.ray;
    self
      .builder
      .execute(move |node_id, dir, etype| ray.neighbors(node_id, dir, etype))
  }

  /// Execute and return iterator over edges only
  ///
  /// This is useful when you want to collect the edges traversed rather than nodes.
  /// Each result contains the source, destination, and edge type of edges encountered.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::NodeId;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let kite: Kite = unimplemented!();
  /// # let user_id: NodeId = 1;
  /// let edges: Vec<_> = kite.from(user_id)
  ///     .out(Some("FOLLOWS"))?
  ///     .edges()
  ///     .collect();
  ///
  /// for edge in edges {
  ///     println!("{} -[{}]-> {}", edge.src, edge.etype, edge.dst);
  /// }
  /// # Ok(())
  /// # }
  /// ```
  pub fn edges(self) -> impl Iterator<Item = Edge> + 'a {
    let ray = self.ray;
    self
      .builder
      .execute(move |node_id, dir, etype| ray.neighbors(node_id, dir, etype))
      .filter_map(|result| {
        result.edge.map(|e| Edge {
          src: e.src,
          etype: e.etype,
          dst: e.dst,
        })
      })
  }

  /// Execute and return iterator over full edge details
  ///
  /// Similar to `edges()` but returns FullEdge structs.
  pub fn full_edges(self) -> impl Iterator<Item = FullEdge> + 'a {
    let ray = self.ray;
    self
      .builder
      .execute(move |node_id, dir, etype| ray.neighbors(node_id, dir, etype))
      .filter_map(move |result| {
        result.edge.map(|e| FullEdge {
          src: e.src,
          etype: e.etype,
          dst: e.dst,
        })
      })
  }

  fn resolve_etype(&self, edge_type: Option<&str>) -> Result<Option<ETypeId>> {
    match edge_type {
      Some(name) => {
        let edge_def = self
          .ray
          .edges
          .get(name)
          .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {name}").into()))?;
        Ok(edge_def.etype_id)
      }
      None => Ok(None),
    }
  }
}

// ============================================================================
// Path Finding Builder for Kite
// ============================================================================

use super::pathfinding::{bfs, dijkstra, yen_k_shortest, PathConfig, PathResult};

/// Path finding builder bound to a Kite database
///
/// Provides ergonomic pathfinding operations using edge type names.
pub struct KitePathBuilder<'a> {
  ray: &'a Kite,
  source: NodeId,
  targets: HashSet<NodeId>,
  allowed_etypes: HashSet<ETypeId>,
  direction: TraversalDirection,
  max_depth: usize,
  weights: HashMap<(NodeId, ETypeId, NodeId), f64>,
}

impl<'a> KitePathBuilder<'a> {
  fn new(ray: &'a Kite, source: NodeId, target: NodeId) -> Self {
    let mut targets = HashSet::new();
    targets.insert(target);

    Self {
      ray,
      source,
      targets,
      allowed_etypes: HashSet::new(),
      direction: TraversalDirection::Out,
      max_depth: 100,
      weights: HashMap::new(),
    }
  }

  fn new_multi(ray: &'a Kite, source: NodeId, targets: Vec<NodeId>) -> Self {
    Self {
      ray,
      source,
      targets: targets.into_iter().collect(),
      allowed_etypes: HashSet::new(),
      direction: TraversalDirection::Out,
      max_depth: 100,
      weights: HashMap::new(),
    }
  }

  /// Restrict traversal to specific edge type
  ///
  /// Can be called multiple times to allow multiple edge types.
  pub fn via(mut self, edge_type: &str) -> Result<Self> {
    let edge_def =
      self.ray.edges.get(edge_type).ok_or_else(|| {
        KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into())
      })?;

    if let Some(etype_id) = edge_def.etype_id {
      self.allowed_etypes.insert(etype_id);
    }

    Ok(self)
  }

  /// Set maximum search depth
  pub fn max_depth(mut self, depth: usize) -> Self {
    self.max_depth = depth;
    self
  }

  /// Set traversal direction
  pub fn direction(mut self, direction: TraversalDirection) -> Self {
    self.direction = direction;
    self
  }

  /// Use bidirectional traversal
  pub fn bidirectional(mut self) -> Self {
    self.direction = TraversalDirection::Both;
    self
  }

  /// Find the shortest path using Dijkstra's algorithm
  pub fn find(self) -> PathResult {
    let config = PathConfig {
      source: self.source,
      targets: self.targets,
      allowed_etypes: self.allowed_etypes,
      direction: self.direction,
      max_depth: self.max_depth,
    };

    let weights = self.weights;
    dijkstra(
      config,
      |node_id, dir, etype| self.ray.neighbors(node_id, dir, etype),
      move |src, etype, dst| weights.get(&(src, etype, dst)).copied().unwrap_or(1.0),
    )
  }

  /// Find the shortest path using BFS (unweighted)
  ///
  /// Faster than Dijkstra for unweighted graphs.
  pub fn find_bfs(self) -> PathResult {
    let config = PathConfig {
      source: self.source,
      targets: self.targets,
      allowed_etypes: self.allowed_etypes,
      direction: self.direction,
      max_depth: self.max_depth,
    };

    bfs(config, |node_id, dir, etype| {
      self.ray.neighbors(node_id, dir, etype)
    })
  }

  /// Find the k shortest paths using Yen's algorithm
  pub fn find_k_shortest(self, k: usize) -> Vec<PathResult> {
    let config = PathConfig {
      source: self.source,
      targets: self.targets,
      allowed_etypes: self.allowed_etypes,
      direction: self.direction,
      max_depth: self.max_depth,
    };

    let weights = self.weights;
    yen_k_shortest(
      config,
      k,
      |node_id, dir, etype| self.ray.neighbors(node_id, dir, etype),
      move |src, etype, dst| weights.get(&(src, etype, dst)).copied().unwrap_or(1.0),
    )
  }
}

// ============================================================================
// Batch Operations
// ============================================================================

/// A batch operation that can be executed atomically with other operations
#[derive(Debug, Clone)]
pub enum BatchOp {
  /// Create a new node
  CreateNode {
    node_type: String,
    key_suffix: String,
    props: HashMap<String, PropValue>,
  },
  /// Delete a node
  DeleteNode { node_id: NodeId },
  /// Create an edge
  Link {
    src: NodeId,
    edge_type: String,
    dst: NodeId,
  },
  /// Create an edge with properties
  LinkWithProps {
    src: NodeId,
    edge_type: String,
    dst: NodeId,
    props: HashMap<String, PropValue>,
  },
  /// Remove an edge
  Unlink {
    src: NodeId,
    edge_type: String,
    dst: NodeId,
  },
  /// Set a node property
  SetProp {
    node_id: NodeId,
    prop_name: String,
    value: PropValue,
  },
  /// Set an edge property
  SetEdgeProp {
    src: NodeId,
    edge_type: String,
    dst: NodeId,
    prop_name: String,
    value: PropValue,
  },
  /// Set multiple edge properties
  SetEdgeProps {
    src: NodeId,
    edge_type: String,
    dst: NodeId,
    props: HashMap<String, PropValue>,
  },
  /// Delete a node property
  DelProp { node_id: NodeId, prop_name: String },
}

/// Result of a batch operation
#[derive(Debug, Clone)]
pub enum BatchResult {
  /// Node was created, contains the NodeRef
  NodeCreated(NodeRef),
  /// Node was deleted
  NodeDeleted(bool),
  /// Edge was created
  EdgeCreated,
  /// Edge was removed
  EdgeRemoved(bool),
  /// Property was set
  PropSet,
  /// Property was deleted
  PropDeleted,
}

#[derive(Debug, Clone)]
struct EdgeCacheEntry {
  etype_id: ETypeId,
  prop_key_ids: HashMap<String, PropKeyId>,
}

fn resolve_edge_cache_entry<'a>(
  edge_cache: &'a mut HashMap<String, EdgeCacheEntry>,
  edges: &HashMap<String, EdgeDef>,
  edge_type: &str,
) -> Result<&'a mut EdgeCacheEntry> {
  if edge_cache.contains_key(edge_type) {
    return Ok(edge_cache.get_mut(edge_type).expect("expected value"));
  }

  let edge_def = edges
    .get(edge_type)
    .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

  let etype_id = edge_def
    .etype_id
    .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

  edge_cache.insert(
    edge_type.to_string(),
    EdgeCacheEntry {
      etype_id,
      prop_key_ids: edge_def.prop_key_ids.clone(),
    },
  );

  Ok(edge_cache.get_mut(edge_type).expect("expected value"))
}

impl Kite {
  /// Execute multiple operations atomically in a single transaction
  ///
  /// All operations succeed or fail together. If any operation fails,
  /// the entire batch is rolled back.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::{BatchOp, Kite, KiteOptions};
  /// # use std::collections::HashMap;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let options = KiteOptions::default();
  /// let mut kite = Kite::open("db", options)?;
  ///
  /// let results = kite.batch(vec![
  ///   BatchOp::CreateNode {
  ///     node_type: "User".into(),
  ///     key_suffix: "alice".into(),
  ///     props: HashMap::new(),
  ///   },
  ///   BatchOp::CreateNode {
  ///     node_type: "User".into(),
  ///     key_suffix: "bob".into(),
  ///     props: HashMap::new(),
  ///   },
  /// ])?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn batch(&mut self, ops: Vec<BatchOp>) -> Result<Vec<BatchResult>> {
    let mut handle = begin_tx(&self.db)?;
    let mut results = Vec::with_capacity(ops.len());
    let mut edge_cache: HashMap<String, EdgeCacheEntry> = HashMap::new();

    for op in ops {
      let result = match op {
        BatchOp::CreateNode {
          node_type,
          key_suffix,
          props,
        } => {
          let node_def = self.nodes.get(&node_type).ok_or_else(|| {
            KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into())
          })?;

          let full_key = node_def.key(&key_suffix);

          let node_opts = NodeOpts {
            key: Some(full_key.clone()),
            labels: node_def.label_id.map(|id| vec![id]),
            props: None,
          };
          let node_id = create_node(&mut handle, node_opts)?;

          // Set properties
          for (prop_name, value) in props {
            if let Some(&prop_key_id) = node_def.prop_key_ids.get(&prop_name) {
              set_node_prop(&mut handle, node_id, prop_key_id, value)?;
            }
          }

          BatchResult::NodeCreated(NodeRef::new(node_id, Some(full_key), node_type))
        }

        BatchOp::DeleteNode { node_id } => {
          let deleted = delete_node(&mut handle, node_id)?;
          BatchResult::NodeDeleted(deleted)
        }

        BatchOp::Link {
          src,
          edge_type,
          dst,
        } => {
          let entry = resolve_edge_cache_entry(&mut edge_cache, &self.edges, &edge_type)?;
          let etype_id = entry.etype_id;
          add_edge(&mut handle, src, etype_id, dst)?;
          BatchResult::EdgeCreated
        }

        BatchOp::LinkWithProps {
          src,
          edge_type,
          dst,
          props,
        } => {
          let entry = resolve_edge_cache_entry(&mut edge_cache, &self.edges, &edge_type)?;
          let etype_id = entry.etype_id;

          if props.is_empty() {
            add_edge(&mut handle, src, etype_id, dst)?;
          } else {
            let mut prop_pairs = Vec::with_capacity(props.len());
            for (prop_name, value) in props {
              let prop_key_id = if let Some(&id) = entry.prop_key_ids.get(&prop_name) {
                id
              } else {
                let key_id = handle.db.propkey_id_or_create(&prop_name);
                entry.prop_key_ids.insert(prop_name.clone(), key_id);
                key_id
              };
              prop_pairs.push((prop_key_id, value));
            }
            handle
              .db
              .add_edge_with_props(src, etype_id, dst, prop_pairs)?;
          }

          BatchResult::EdgeCreated
        }

        BatchOp::Unlink {
          src,
          edge_type,
          dst,
        } => {
          let edge_def = self.edges.get(&edge_type).ok_or_else(|| {
            KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into())
          })?;

          let etype_id = edge_def
            .etype_id
            .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

          let deleted = delete_edge(&mut handle, src, etype_id, dst)?;
          BatchResult::EdgeRemoved(deleted)
        }

        BatchOp::SetProp {
          node_id,
          prop_name,
          value,
        } => {
          // Use handle.db to access schema methods while handle is active
          let prop_key_id = handle.db.propkey_id_or_create(&prop_name);
          set_node_prop(&mut handle, node_id, prop_key_id, value)?;
          BatchResult::PropSet
        }

        BatchOp::SetEdgeProp {
          src,
          edge_type,
          dst,
          prop_name,
          value,
        } => {
          let entry = resolve_edge_cache_entry(&mut edge_cache, &self.edges, &edge_type)?;
          let etype_id = entry.etype_id;

          let prop_key_id = if let Some(&id) = entry.prop_key_ids.get(&prop_name) {
            id
          } else {
            let key_id = handle.db.propkey_id_or_create(&prop_name);
            entry.prop_key_ids.insert(prop_name.clone(), key_id);
            key_id
          };

          set_edge_prop(&mut handle, src, etype_id, dst, prop_key_id, value)?;
          BatchResult::PropSet
        }

        BatchOp::SetEdgeProps {
          src,
          edge_type,
          dst,
          props,
        } => {
          let entry = resolve_edge_cache_entry(&mut edge_cache, &self.edges, &edge_type)?;
          let etype_id = entry.etype_id;

          let mut prop_pairs = Vec::with_capacity(props.len());
          for (prop_name, value) in props {
            let prop_key_id = if let Some(&id) = entry.prop_key_ids.get(&prop_name) {
              id
            } else {
              let key_id = handle.db.propkey_id_or_create(&prop_name);
              entry.prop_key_ids.insert(prop_name.clone(), key_id);
              key_id
            };
            prop_pairs.push((prop_key_id, value));
          }

          handle.db.set_edge_props(src, etype_id, dst, prop_pairs)?;
          BatchResult::PropSet
        }

        BatchOp::DelProp { node_id, prop_name } => {
          let prop_key_id = handle.db.propkey_id(&prop_name).ok_or_else(|| {
            KiteError::InvalidSchema(format!("Unknown property: {prop_name}").into())
          })?;
          del_node_prop(&mut handle, node_id, prop_key_id)?;
          BatchResult::PropDeleted
        }
      };

      results.push(result);
    }

    // Commit the entire batch
    commit(&mut handle)?;

    Ok(results)
  }
}

// ============================================================================
// Transaction Context
// ============================================================================

/// Context for executing operations within a transaction
///
/// Provides the same operations as Kite but within an explicit transaction scope.
/// All operations are committed together when the transaction closure returns Ok,
/// or rolled back if an error is returned.
///
/// Note: TxContext holds references to the schema maps (nodes, edges) separately
/// from the TxHandle to avoid borrow checker issues.
pub struct TxContext<'a> {
  handle: TxHandle<'a>,
  nodes: &'a HashMap<String, NodeDef>,
  edges: &'a HashMap<String, EdgeDef>,
}

impl<'a> TxContext<'a> {
  /// Create a new node
  pub fn create_node(
    &mut self,
    node_type: &str,
    key_suffix: &str,
    props: HashMap<String, PropValue>,
  ) -> Result<NodeRef> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?
      .clone();

    let full_key = node_def.key(key_suffix);

    let node_opts = NodeOpts {
      key: Some(full_key.clone()),
      labels: node_def.label_id.map(|id| vec![id]),
      props: None,
    };
    let node_id = create_node(&mut self.handle, node_opts)?;

    // Set properties
    for (prop_name, value) in props {
      if let Some(&prop_key_id) = node_def.prop_key_ids.get(&prop_name) {
        set_node_prop(&mut self.handle, node_id, prop_key_id, value)?;
      }
    }

    Ok(NodeRef::new(node_id, Some(full_key), node_type))
  }

  /// Delete a node
  pub fn delete_node(&mut self, node_id: NodeId) -> Result<bool> {
    delete_node(&mut self.handle, node_id)
  }

  /// Create an edge
  pub fn link(&mut self, src: NodeId, edge_type: &str, dst: NodeId) -> Result<()> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    add_edge(&mut self.handle, src, etype_id, dst)?;
    Ok(())
  }

  /// Remove an edge
  pub fn unlink(&mut self, src: NodeId, edge_type: &str, dst: NodeId) -> Result<bool> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    delete_edge(&mut self.handle, src, etype_id, dst)
  }

  /// Set a node property
  pub fn set_prop(&mut self, node_id: NodeId, prop_name: &str, value: PropValue) -> Result<()> {
    let prop_key_id = self.handle.db.propkey_id_or_create(prop_name);
    set_node_prop(&mut self.handle, node_id, prop_key_id, value)?;
    Ok(())
  }

  /// Delete a node property
  pub fn del_prop(&mut self, node_id: NodeId, prop_name: &str) -> Result<()> {
    let prop_key_id = self
      .handle
      .db
      .propkey_id(prop_name)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown property: {prop_name}").into()))?;
    del_node_prop(&mut self.handle, node_id, prop_key_id)?;
    Ok(())
  }

  /// Check if a node exists
  pub fn exists(&self, node_id: NodeId) -> bool {
    node_exists(&self.handle, node_id)
  }

  /// Check if an edge exists
  pub fn has_edge(&self, src: NodeId, edge_type: &str, dst: NodeId) -> Result<bool> {
    let edge_def = self
      .edges
      .get(edge_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown edge type: {edge_type}").into()))?;

    let etype_id = edge_def
      .etype_id
      .ok_or_else(|| KiteError::InvalidSchema("Edge type not initialized".into()))?;

    Ok(edge_exists(&self.handle, src, etype_id, dst))
  }

  /// Get a node property
  pub fn prop(&self, node_id: NodeId, prop_name: &str) -> Result<Option<PropValue>> {
    let prop_key_id = self
      .handle
      .db
      .propkey_id(prop_name)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown property: {prop_name}").into()))?;

    Ok(node_prop(&self.handle, node_id, prop_key_id))
  }

  /// Get a node by key
  pub fn get(&self, node_type: &str, key_suffix: &str) -> Result<Option<NodeRef>> {
    let node_def = self
      .nodes
      .get(node_type)
      .ok_or_else(|| KiteError::InvalidSchema(format!("Unknown node type: {node_type}").into()))?;

    let full_key = node_def.key(key_suffix);
    let node_id = node_by_key(&self.handle, &full_key);

    match node_id {
      Some(id) => Ok(Some(NodeRef::new(id, Some(full_key), node_type))),
      None => Ok(None),
    }
  }
}

impl Kite {
  /// Execute operations in an explicit transaction
  ///
  /// The closure receives a TxContext with access to node/edge operations.
  /// All operations performed through the context are committed together when
  /// the closure returns Ok, or rolled back if an error is returned.
  ///
  /// # Example
  /// ```rust,no_run
  /// # use kitedb::api::kite::Kite;
  /// # use kitedb::types::PropValue;
  /// # use std::collections::HashMap;
  /// # fn main() -> kitedb::error::Result<()> {
  /// # let mut kite: Kite = unimplemented!();
  /// let result = kite.transaction(|ctx| {
  ///   let alice = ctx.create_node("User", "alice", HashMap::new())?;
  ///   let bob = ctx.create_node("User", "bob", HashMap::new())?;
  ///   ctx.link(alice.id(), "FOLLOWS", bob.id())?;
  ///   Ok((alice, bob))
  /// })?;
  /// # Ok(())
  /// # }
  /// ```
  pub fn transaction<T, F>(&mut self, f: F) -> Result<T>
  where
    F: FnOnce(&mut TxContext) -> Result<T>,
  {
    // Start the transaction
    let handle = begin_tx(&self.db)?;

    // Create context with references to schema maps
    let mut ctx = TxContext {
      handle,
      nodes: &self.nodes,
      edges: &self.edges,
    };

    match f(&mut ctx) {
      Ok(result) => {
        commit(&mut ctx.handle)?;
        Ok(result)
      }
      Err(e) => {
        rollback(&mut ctx.handle)?;
        Err(e)
      }
    }
  }

  /// Execute a transaction with a simpler API using a builder pattern
  ///
  /// Returns a TxBuilder that collects operations and executes them atomically.
  pub fn tx(&mut self) -> TxBuilder {
    TxBuilder { ops: Vec::new() }
  }
}

/// Builder for constructing transactions with a fluent API
#[derive(Debug, Default)]
pub struct TxBuilder {
  ops: Vec<BatchOp>,
}

impl TxBuilder {
  /// Add a create node operation
  pub fn create_node(
    mut self,
    node_type: impl Into<String>,
    key_suffix: impl Into<String>,
    props: HashMap<String, PropValue>,
  ) -> Self {
    self.ops.push(BatchOp::CreateNode {
      node_type: node_type.into(),
      key_suffix: key_suffix.into(),
      props,
    });
    self
  }

  /// Add a delete node operation
  pub fn delete_node(mut self, node_id: NodeId) -> Self {
    self.ops.push(BatchOp::DeleteNode { node_id });
    self
  }

  /// Add a link operation
  pub fn link(mut self, src: NodeId, edge_type: impl Into<String>, dst: NodeId) -> Self {
    self.ops.push(BatchOp::Link {
      src,
      edge_type: edge_type.into(),
      dst,
    });
    self
  }

  /// Add a link operation with properties
  pub fn link_with_props(
    mut self,
    src: NodeId,
    edge_type: impl Into<String>,
    dst: NodeId,
    props: HashMap<String, PropValue>,
  ) -> Self {
    self.ops.push(BatchOp::LinkWithProps {
      src,
      edge_type: edge_type.into(),
      dst,
      props,
    });
    self
  }

  /// Add an unlink operation
  pub fn unlink(mut self, src: NodeId, edge_type: impl Into<String>, dst: NodeId) -> Self {
    self.ops.push(BatchOp::Unlink {
      src,
      edge_type: edge_type.into(),
      dst,
    });
    self
  }

  /// Add a set property operation
  pub fn set_prop(
    mut self,
    node_id: NodeId,
    prop_name: impl Into<String>,
    value: PropValue,
  ) -> Self {
    self.ops.push(BatchOp::SetProp {
      node_id,
      prop_name: prop_name.into(),
      value,
    });
    self
  }

  /// Add a set edge property operation
  pub fn set_edge_prop(
    mut self,
    src: NodeId,
    edge_type: impl Into<String>,
    dst: NodeId,
    prop_name: impl Into<String>,
    value: PropValue,
  ) -> Self {
    self.ops.push(BatchOp::SetEdgeProp {
      src,
      edge_type: edge_type.into(),
      dst,
      prop_name: prop_name.into(),
      value,
    });
    self
  }

  /// Add a set edge properties operation
  pub fn set_edge_props(
    mut self,
    src: NodeId,
    edge_type: impl Into<String>,
    dst: NodeId,
    props: HashMap<String, PropValue>,
  ) -> Self {
    self.ops.push(BatchOp::SetEdgeProps {
      src,
      edge_type: edge_type.into(),
      dst,
      props,
    });
    self
  }

  /// Add a delete property operation
  pub fn del_prop(mut self, node_id: NodeId, prop_name: impl Into<String>) -> Self {
    self.ops.push(BatchOp::DelProp {
      node_id,
      prop_name: prop_name.into(),
    });
    self
  }

  /// Execute the transaction on the given Kite instance
  pub fn execute(self, ray: &mut Kite) -> Result<Vec<BatchResult>> {
    ray.batch(self.ops)
  }

  /// Get the operations as a Vec<BatchOp>
  pub fn into_ops(self) -> Vec<BatchOp> {
    self.ops
  }
}

// ============================================================================
// Update Node Builder
// ============================================================================

/// Fluent builder for updating node properties
///
/// Created via `kite.update()`, `kite.update_by_id()`, or `kite.update_by_key()`
/// and allows chaining multiple property set/unset operations before executing
/// in a single transaction.
///
/// # Example
/// ```rust,no_run
/// # use kitedb::api::kite::{NodeRef, Kite};
/// # use kitedb::types::PropValue;
/// # fn main() -> kitedb::error::Result<()> {
/// # let mut kite: Kite = unimplemented!();
/// # let alice: NodeRef = unimplemented!();
/// // Update by node reference
/// kite.update(&alice)?
///     .set("name", PropValue::String("Alice Updated".into()))
///     .set("age", PropValue::I64(31))
///     .unset("old_field")
///     .execute()?;
///
/// // Update by key
/// kite.update_by_key("User", "alice")?
///     .set("name", PropValue::String("New Name".into()))
///     .execute()?;
/// # Ok(())
/// # }
/// ```
pub struct KiteUpdateNodeBuilder<'a> {
  ray: &'a mut Kite,
  node_id: NodeId,
  updates: HashMap<String, Option<PropValue>>,
}

impl<'a> KiteUpdateNodeBuilder<'a> {
  /// Set a node property value
  ///
  /// The property will be set when `execute()` is called.
  pub fn set(mut self, prop_name: impl Into<String>, value: PropValue) -> Self {
    self.updates.insert(prop_name.into(), Some(value));
    self
  }

  /// Remove a node property
  ///
  /// The property will be deleted when `execute()` is called.
  pub fn unset(mut self, prop_name: impl Into<String>) -> Self {
    self.updates.insert(prop_name.into(), None);
    self
  }

  /// Set multiple properties at once from a HashMap
  ///
  /// Convenience method for setting multiple properties.
  pub fn set_all(mut self, props: HashMap<String, PropValue>) -> Self {
    for (k, v) in props {
      self.updates.insert(k, Some(v));
    }
    self
  }

  /// Execute the update, applying all property changes in a single transaction
  pub fn execute(self) -> Result<()> {
    if self.updates.is_empty() {
      return Ok(());
    }

    let mut handle = begin_tx(&self.ray.db)?;

    for (prop_name, value_opt) in self.updates {
      let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);

      match value_opt {
        Some(value) => {
          set_node_prop(&mut handle, self.node_id, prop_key_id, value)?;
        }
        None => {
          // Only delete if prop exists
          del_node_prop(&mut handle, self.node_id, prop_key_id)?;
        }
      }
    }

    commit(&mut handle)?;
    Ok(())
  }

  /// Get the node ID being updated
  pub fn node_id(&self) -> NodeId {
    self.node_id
  }
}

// ============================================================================
// Upsert By ID Builder
// ============================================================================

/// Fluent builder for upserting a node by ID
///
/// Created via `kite.upsert_by_id(node_type, node_id)` and allows chaining
/// property set/unset operations before executing in a single transaction.
pub struct KiteUpsertByIdBuilder<'a> {
  ray: &'a mut Kite,
  node_id: NodeId,
  node_def: NodeDef,
  updates: HashMap<String, Option<PropValue>>,
}

impl<'a> KiteUpsertByIdBuilder<'a> {
  /// Set a node property value
  pub fn set(mut self, prop_name: impl Into<String>, value: PropValue) -> Self {
    self.updates.insert(prop_name.into(), Some(value));
    self
  }

  /// Remove a node property
  pub fn unset(mut self, prop_name: impl Into<String>) -> Self {
    self.updates.insert(prop_name.into(), None);
    self
  }

  /// Set multiple properties at once from a HashMap
  pub fn set_all(mut self, props: HashMap<String, PropValue>) -> Self {
    for (k, v) in props {
      self.updates.insert(k, Some(v));
    }
    self
  }

  /// Execute the upsert, creating the node if missing
  pub fn execute(self) -> Result<()> {
    let mut handle = begin_tx(&self.ray.db)?;

    let mut updates = Vec::with_capacity(self.updates.len());
    for (prop_name, value_opt) in self.updates {
      let prop_key_id = if let Some(&id) = self.node_def.prop_key_ids.get(&prop_name) {
        id
      } else {
        self.ray.db.propkey_id_or_create(&prop_name)
      };
      updates.push((prop_key_id, value_opt));
    }

    let opts = NodeOpts {
      key: None,
      labels: self.node_def.label_id.map(|id| vec![id]),
      props: None,
    };

    upsert_node_by_id_with_props(&mut handle, self.node_id, opts, updates)?;

    commit(&mut handle)?;
    Ok(())
  }
}

// ============================================================================
// Insert Builder
// ============================================================================

/// Fluent builder for inserting nodes
///
/// Created via `kite.insert(node_type)` and provides a fluent API for creating
/// nodes with the `.values().returning()` or `.values().execute()` pattern.
///
/// # Example
/// ```rust,no_run
/// # use kitedb::api::kite::Kite;
/// # use kitedb::types::PropValue;
/// # use std::collections::HashMap;
/// # fn main() -> kitedb::error::Result<()> {
/// # let mut kite: Kite = unimplemented!();
/// # let props: HashMap<String, PropValue> = HashMap::new();
/// # let alice_props: HashMap<String, PropValue> = HashMap::new();
/// # let bob_props: HashMap<String, PropValue> = HashMap::new();
/// // Insert and get the node reference back
/// let user = kite.insert("User")?
///     .values("alice", props)?
///     .returning()?;
///
/// // Insert multiple nodes
/// let users = kite.insert("User")?
///     .values_many(vec![
///         ("alice", alice_props),
///         ("bob", bob_props),
///     ])?
///     .returning()?;
/// # Ok(())
/// # }
/// ```
pub struct KiteInsertBuilder<'a> {
  ray: &'a mut Kite,
  node_type: String,
  key_prefix: String,
}

impl<'a> KiteInsertBuilder<'a> {
  /// Specify the values for a single node insert
  ///
  /// Returns an executor that can either `.execute()` (no return) or
  /// `.returning()` (returns NodeRef).
  pub fn values(
    self,
    key_suffix: &str,
    props: HashMap<String, PropValue>,
  ) -> Result<InsertExecutorSingle<'a>> {
    let full_key = format!("{}{}", self.key_prefix, key_suffix);
    Ok(InsertExecutorSingle {
      ray: self.ray,
      node_type: self.node_type,
      full_key,
      props,
    })
  }

  /// Specify values for multiple nodes
  ///
  /// Returns an executor that can either `.execute()` (no return) or
  /// `.returning()` (returns Vec<NodeRef>).
  pub fn values_many(
    self,
    items: Vec<(&str, HashMap<String, PropValue>)>,
  ) -> Result<InsertExecutorMultiple<'a>> {
    let entries: Vec<(String, HashMap<String, PropValue>)> = items
      .into_iter()
      .map(|(key_suffix, props)| {
        let full_key = format!("{}{}", self.key_prefix, key_suffix);
        (full_key, props)
      })
      .collect();

    Ok(InsertExecutorMultiple {
      ray: self.ray,
      node_type: self.node_type,
      entries,
    })
  }

  /// Specify values for multiple nodes with owned key suffixes
  pub fn values_many_owned(
    self,
    items: Vec<(String, HashMap<String, PropValue>)>,
  ) -> Result<InsertExecutorMultiple<'a>> {
    let entries: Vec<(String, HashMap<String, PropValue>)> = items
      .into_iter()
      .map(|(key_suffix, props)| {
        let full_key = format!("{}{}", self.key_prefix, key_suffix);
        (full_key, props)
      })
      .collect();

    Ok(InsertExecutorMultiple {
      ray: self.ray,
      node_type: self.node_type,
      entries,
    })
  }
}

/// Executor for single node insert
pub struct InsertExecutorSingle<'a> {
  ray: &'a mut Kite,
  node_type: String,
  full_key: String,
  props: HashMap<String, PropValue>,
}

impl<'a> InsertExecutorSingle<'a> {
  /// Execute the insert and return the created node reference
  pub fn returning(self) -> Result<NodeRef> {
    let node_type: Arc<str> = self.node_type.into();
    let mut handle = begin_tx(&self.ray.db)?;

    // Create the node
    let node_opts = NodeOpts::new().with_key(self.full_key.clone());
    let node_id = create_node(&mut handle, node_opts)?;

    // Set properties
    for (prop_name, value) in self.props {
      let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);
      set_node_prop(&mut handle, node_id, prop_key_id, value)?;
    }

    commit(&mut handle)?;

    Ok(NodeRef::new(node_id, Some(self.full_key), node_type))
  }

  /// Execute the insert without returning the node reference
  ///
  /// Slightly more efficient when you don't need the result.
  pub fn execute(self) -> Result<()> {
    let _ = self.returning()?;
    Ok(())
  }
}

/// Executor for multiple node insert
pub struct InsertExecutorMultiple<'a> {
  ray: &'a mut Kite,
  node_type: String,
  entries: Vec<(String, HashMap<String, PropValue>)>,
}

impl<'a> InsertExecutorMultiple<'a> {
  /// Execute the insert and return all created node references
  pub fn returning(self) -> Result<Vec<NodeRef>> {
    if self.entries.is_empty() {
      return Ok(Vec::new());
    }

    let mut handle = begin_tx(&self.ray.db)?;
    let mut results = Vec::with_capacity(self.entries.len());
    let node_type: Arc<str> = self.node_type.into();

    for (full_key, props) in self.entries {
      // Create the node
      let node_opts = NodeOpts::new().with_key(full_key.clone());
      let node_id = create_node(&mut handle, node_opts)?;

      // Set properties
      for (prop_name, value) in props {
        let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);
        set_node_prop(&mut handle, node_id, prop_key_id, value)?;
      }

      results.push(NodeRef::new(
        node_id,
        Some(full_key),
        Arc::clone(&node_type),
      ));
    }

    commit(&mut handle)?;

    Ok(results)
  }

  /// Execute the insert without returning node references
  pub fn execute(self) -> Result<()> {
    let _ = self.returning()?;
    Ok(())
  }
}

// ============================================================================
// Upsert Builder
// ============================================================================

/// Fluent builder for upserting nodes
///
/// Created via `kite.upsert(node_type)` and provides a fluent API for
/// creating or updating nodes with the `.values().returning()` or
/// `.values().execute()` pattern.
pub struct KiteUpsertBuilder<'a> {
  ray: &'a mut Kite,
  node_type: String,
  key_prefix: String,
}

impl<'a> KiteUpsertBuilder<'a> {
  /// Specify the values for a single upsert
  pub fn values(
    self,
    key_suffix: &str,
    props: HashMap<String, PropValue>,
  ) -> Result<UpsertExecutorSingle<'a>> {
    let full_key = format!("{}{}", self.key_prefix, key_suffix);
    Ok(UpsertExecutorSingle {
      ray: self.ray,
      node_type: self.node_type,
      full_key,
      props,
    })
  }

  /// Specify values for multiple upserts
  pub fn values_many(
    self,
    items: Vec<(&str, HashMap<String, PropValue>)>,
  ) -> Result<UpsertExecutorMultiple<'a>> {
    let entries: Vec<(String, HashMap<String, PropValue>)> = items
      .into_iter()
      .map(|(key_suffix, props)| {
        let full_key = format!("{}{}", self.key_prefix, key_suffix);
        (full_key, props)
      })
      .collect();

    Ok(UpsertExecutorMultiple {
      ray: self.ray,
      node_type: self.node_type,
      entries,
    })
  }

  /// Specify values for multiple upserts with owned key suffixes
  pub fn values_many_owned(
    self,
    items: Vec<(String, HashMap<String, PropValue>)>,
  ) -> Result<UpsertExecutorMultiple<'a>> {
    let entries: Vec<(String, HashMap<String, PropValue>)> = items
      .into_iter()
      .map(|(key_suffix, props)| {
        let full_key = format!("{}{}", self.key_prefix, key_suffix);
        (full_key, props)
      })
      .collect();

    Ok(UpsertExecutorMultiple {
      ray: self.ray,
      node_type: self.node_type,
      entries,
    })
  }
}

/// Executor for single node upsert
pub struct UpsertExecutorSingle<'a> {
  ray: &'a mut Kite,
  node_type: String,
  full_key: String,
  props: HashMap<String, PropValue>,
}

impl<'a> UpsertExecutorSingle<'a> {
  /// Execute the upsert and return the node reference
  pub fn returning(self) -> Result<NodeRef> {
    let node_type: Arc<str> = self.node_type.into();
    let mut handle = begin_tx(&self.ray.db)?;

    let mut updates = Vec::with_capacity(self.props.len());
    for (prop_name, value) in self.props {
      let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);
      let value_opt = match value {
        PropValue::Null => None,
        other => Some(other),
      };
      updates.push((prop_key_id, value_opt));
    }

    let (node_id, _) = upsert_node_with_props(&mut handle, &self.full_key, updates)?;

    commit(&mut handle)?;

    Ok(NodeRef::new(node_id, Some(self.full_key), node_type))
  }

  /// Execute the upsert without returning the node reference
  pub fn execute(self) -> Result<()> {
    let _ = self.returning()?;
    Ok(())
  }
}

/// Executor for multiple node upserts
pub struct UpsertExecutorMultiple<'a> {
  ray: &'a mut Kite,
  node_type: String,
  entries: Vec<(String, HashMap<String, PropValue>)>,
}

impl<'a> UpsertExecutorMultiple<'a> {
  /// Execute the upserts and return node references
  pub fn returning(self) -> Result<Vec<NodeRef>> {
    if self.entries.is_empty() {
      return Ok(Vec::new());
    }

    let mut handle = begin_tx(&self.ray.db)?;
    let mut results = Vec::with_capacity(self.entries.len());
    let node_type: Arc<str> = self.node_type.into();

    for (full_key, props) in self.entries {
      let mut updates = Vec::with_capacity(props.len());
      for (prop_name, value) in props {
        let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);
        let value_opt = match value {
          PropValue::Null => None,
          other => Some(other),
        };
        updates.push((prop_key_id, value_opt));
      }

      let (node_id, _) = upsert_node_with_props(&mut handle, &full_key, updates)?;
      results.push(NodeRef::new(
        node_id,
        Some(full_key),
        Arc::clone(&node_type),
      ));
    }

    commit(&mut handle)?;

    Ok(results)
  }

  /// Execute the upserts without returning node references
  pub fn execute(self) -> Result<()> {
    let _ = self.returning()?;
    Ok(())
  }
}

// ============================================================================
// Update Edge Builder
// ============================================================================

/// Fluent builder for updating edge properties
///
/// Created via `kite.update_edge(src, edge_type, dst)` and allows chaining
/// multiple property set/unset operations before executing in a single transaction.
///
/// # Example
/// ```rust,no_run
/// # use kitedb::api::kite::Kite;
/// # use kitedb::types::{NodeId, PropValue};
/// # fn main() -> kitedb::error::Result<()> {
/// # let mut kite: Kite = unimplemented!();
/// # let alice_id: NodeId = 1;
/// # let bob_id: NodeId = 2;
/// kite.update_edge(alice_id, "FOLLOWS", bob_id)?
///    .set("weight", PropValue::F64(0.9))
///    .set("since", PropValue::String("2024".to_string()))
///    .unset("deprecated_field")
///    .execute()?;
/// # Ok(())
/// # }
/// ```
pub struct KiteUpdateEdgeBuilder<'a> {
  ray: &'a mut Kite,
  src: NodeId,
  etype_id: ETypeId,
  dst: NodeId,
  updates: HashMap<String, Option<PropValue>>,
}

impl<'a> KiteUpdateEdgeBuilder<'a> {
  /// Set an edge property value
  ///
  /// The property will be set when `execute()` is called.
  pub fn set(mut self, prop_name: impl Into<String>, value: PropValue) -> Self {
    self.updates.insert(prop_name.into(), Some(value));
    self
  }

  /// Remove an edge property
  ///
  /// The property will be deleted when `execute()` is called.
  pub fn unset(mut self, prop_name: impl Into<String>) -> Self {
    self.updates.insert(prop_name.into(), None);
    self
  }

  /// Set multiple properties at once from a HashMap
  ///
  /// Convenience method for setting multiple properties.
  pub fn set_all(mut self, props: HashMap<String, PropValue>) -> Self {
    for (k, v) in props {
      self.updates.insert(k, Some(v));
    }
    self
  }

  /// Execute the update, applying all property changes in a single transaction
  pub fn execute(self) -> Result<()> {
    if self.updates.is_empty() {
      return Ok(());
    }

    let mut handle = begin_tx(&self.ray.db)?;

    for (prop_name, value_opt) in self.updates {
      let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);

      match value_opt {
        Some(value) => {
          set_edge_prop(
            &mut handle,
            self.src,
            self.etype_id,
            self.dst,
            prop_key_id,
            value,
          )?;
        }
        None => {
          // Only delete if prop_key exists
          if let Some(existing_key_id) = self.ray.db.propkey_id(&prop_name) {
            del_edge_prop(
              &mut handle,
              self.src,
              self.etype_id,
              self.dst,
              existing_key_id,
            )?;
          }
        }
      }
    }

    commit(&mut handle)?;
    Ok(())
  }
}

// ============================================================================
// Upsert Edge Builder
// ============================================================================

/// Fluent builder for upserting edge properties
///
/// Created via `kite.upsert_edge(src, edge_type, dst)` and allows chaining
/// property set/unset operations before executing in a single transaction.
pub struct KiteUpsertEdgeBuilder<'a> {
  ray: &'a mut Kite,
  src: NodeId,
  etype_id: ETypeId,
  dst: NodeId,
  updates: HashMap<String, Option<PropValue>>,
}

impl<'a> KiteUpsertEdgeBuilder<'a> {
  /// Set an edge property value
  pub fn set(mut self, prop_name: impl Into<String>, value: PropValue) -> Self {
    self.updates.insert(prop_name.into(), Some(value));
    self
  }

  /// Remove an edge property
  pub fn unset(mut self, prop_name: impl Into<String>) -> Self {
    self.updates.insert(prop_name.into(), None);
    self
  }

  /// Set multiple properties at once from a HashMap
  pub fn set_all(mut self, props: HashMap<String, PropValue>) -> Self {
    for (k, v) in props {
      self.updates.insert(k, Some(v));
    }
    self
  }

  /// Execute the upsert, creating the edge if missing
  pub fn execute(self) -> Result<()> {
    let mut handle = begin_tx(&self.ray.db)?;

    let mut updates = Vec::with_capacity(self.updates.len());
    for (prop_name, value_opt) in self.updates {
      let prop_key_id = self.ray.db.propkey_id_or_create(&prop_name);
      updates.push((prop_key_id, value_opt));
    }

    upsert_edge_with_props(&mut handle, self.src, self.etype_id, self.dst, updates)?;

    commit(&mut handle)?;
    Ok(())
  }
}

// ============================================================================
// Tests
// ============================================================================

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

  fn temp_db_path(temp_dir: &tempfile::TempDir) -> std::path::PathBuf {
    temp_dir.path().join("test-db")
  }

  fn create_test_schema() -> KiteOptions {
    let user = NodeDef::new("User", "user:")
      .prop(PropDef::string("name").required())
      .prop(PropDef::int("age"));

    let post = NodeDef::new("Post", "post:")
      .prop(PropDef::string("title").required())
      .prop(PropDef::string("content"));

    let follows = EdgeDef::new("FOLLOWS");
    let authored = EdgeDef::new("AUTHORED");

    KiteOptions::new()
      .node(user)
      .node(post)
      .edge(follows)
      .edge(authored)
  }

  #[test]
  fn test_open_database() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    assert_eq!(ray.node_types().len(), 2);
    assert_eq!(ray.edge_types().len(), 2);
    assert!(ray.node_def("User").is_some());
    assert!(ray.edge_def("FOLLOWS").is_some());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_create_and_find_node() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a user
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Alice".to_string()));
    props.insert("age".to_string(), PropValue::I64(30));

    let user_ref = ray
      .create_node("User", "alice", props)
      .expect("expected value");
    assert!(user_ref.id > 0);
    assert_eq!(user_ref.key, Some("user:alice".to_string()));

    // Get the user
    let found = ray.get("User", "alice").expect("expected value");
    assert!(found.is_some());
    assert_eq!(found.expect("expected value").id, user_ref.id);

    // Non-existent user
    let not_found = ray.get("User", "bob").expect("expected value");
    assert!(not_found.is_none());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_link_and_unlink() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create two users
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Link them
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Check edge exists
    assert!(ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));
    assert!(!ray
      .has_edge(bob.id, "FOLLOWS", alice.id)
      .expect("expected value"));

    // Check neighbors
    let alice_follows = ray
      .neighbors_out(alice.id, Some("FOLLOWS"))
      .expect("expected value");
    assert_eq!(alice_follows, vec![bob.id]);

    let bob_followers = ray
      .neighbors_in(bob.id, Some("FOLLOWS"))
      .expect("expected value");
    assert_eq!(bob_followers, vec![alice.id]);

    // Unlink
    ray
      .unlink(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    assert!(!ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_properties() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a user
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Alice".to_string()));
    let user = ray
      .create_node("User", "alice", props)
      .expect("expected value");

    // Get property
    let name = ray.prop(user.id, "name");
    assert_eq!(name, Some(PropValue::String("Alice".to_string())));

    // Set property
    ray
      .set_prop(user.id, "age", PropValue::I64(25))
      .expect("expected value");
    let age = ray.prop(user.id, "age");
    assert_eq!(age, Some(PropValue::I64(25)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_count_nodes() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    assert_eq!(ray.count_nodes(), 0);

    ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .create_node("Post", "post1", HashMap::new())
      .expect("expected value");

    assert_eq!(ray.count_nodes(), 3);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_delete_node() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let user = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    assert!(ray.exists(user.id));

    ray.delete_node(user.id).expect("expected value");
    assert!(!ray.exists(user.id));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_ref() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a user
    let user = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");

    // Get lightweight reference
    let node_ref = ray.node_ref("User", "alice").expect("expected value");
    assert!(node_ref.is_some());
    let node_ref = node_ref.expect("expected value");
    assert_eq!(node_ref.id(), user.id);
    assert_eq!(node_ref.key(), Some("user:alice"));

    // Non-existent user
    let not_found = ray.node_ref("User", "bob").expect("expected value");
    assert!(not_found.is_none());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_all_nodes_by_type() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create some users and posts
    ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .create_node("Post", "post1", HashMap::new())
      .expect("expected value");

    // Iterate all users
    let users: Vec<_> = ray.all("User").expect("expected value").collect();
    assert_eq!(users.len(), 2);

    // Iterate all posts
    let posts: Vec<_> = ray.all("Post").expect("expected value").collect();
    assert_eq!(posts.len(), 1);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_count_nodes_by_type() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create some users and posts
    ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .create_node("Post", "post1", HashMap::new())
      .expect("expected value");

    // Count by type
    assert_eq!(ray.count_nodes_by_type("User").expect("expected value"), 2);
    assert_eq!(ray.count_nodes_by_type("Post").expect("expected value"), 1);
    assert_eq!(ray.count_nodes(), 3);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_all_edges() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create nodes and edges
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let post = ray
      .create_node("Post", "post1", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(alice.id, "AUTHORED", post.id)
      .expect("expected value");

    // List all edges
    let all_edges: Vec<_> = ray.all_edges(None).expect("expected value").collect();
    assert_eq!(all_edges.len(), 2);

    // List FOLLOWS edges only
    let follows_edges: Vec<_> = ray
      .all_edges(Some("FOLLOWS"))
      .expect("expected value")
      .collect();
    assert_eq!(follows_edges.len(), 1);
    assert_eq!(follows_edges[0].src, alice.id);
    assert_eq!(follows_edges[0].dst, bob.id);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_count_edges_by_type() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let post = ray
      .create_node("Post", "post1", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(alice.id, "AUTHORED", post.id)
      .expect("expected value");

    // Count by type
    assert_eq!(
      ray.count_edges_by_type("FOLLOWS").expect("expected value"),
      1
    );
    assert_eq!(
      ray.count_edges_by_type("AUTHORED").expect("expected value"),
      1
    );
    assert_eq!(ray.count_edges(), 2);

    ray.close().expect("expected value");
  }

  // ============================================================================
  // Traversal Tests
  // ============================================================================

  #[test]
  fn test_from_traversal() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a chain: alice -> bob -> charlie
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(bob.id, "FOLLOWS", charlie.id)
      .expect("expected value");

    // Single hop traversal
    let friends = ray
      .from(alice.id)
      .out(Some("FOLLOWS"))
      .expect("expected value")
      .to_vec();
    assert_eq!(friends, vec![bob.id]);

    // Two hop traversal
    let friends_of_friends = ray
      .from(alice.id)
      .out(Some("FOLLOWS"))
      .expect("expected value")
      .out(Some("FOLLOWS"))
      .expect("expected value")
      .to_vec();
    assert_eq!(friends_of_friends, vec![charlie.id]);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_traversal_first() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Get first result
    let first = ray
      .from(alice.id)
      .out(Some("FOLLOWS"))
      .expect("expected value")
      .first_node();
    assert_eq!(first, Some(bob.id));

    // No results
    let no_result = ray
      .from(bob.id)
      .out(Some("FOLLOWS"))
      .expect("expected value")
      .first_node();
    assert_eq!(no_result, None);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_traversal_count() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", charlie.id)
      .expect("expected value");

    let count = ray
      .from(alice.id)
      .out(Some("FOLLOWS"))
      .expect("expected value")
      .count();
    assert_eq!(count, 2);

    ray.close().expect("expected value");
  }

  // ============================================================================
  // Pathfinding Tests
  // ============================================================================

  #[test]
  fn test_shortest_path() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a chain: alice -> bob -> charlie
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(bob.id, "FOLLOWS", charlie.id)
      .expect("expected value");

    // Find path
    let path = ray
      .shortest_path(alice.id, charlie.id)
      .via("FOLLOWS")
      .expect("expected value")
      .find();

    assert!(path.found);
    assert_eq!(path.path, vec![alice.id, bob.id, charlie.id]);
    assert_eq!(path.edges.len(), 2);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_shortest_path_not_found() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // No edge between them
    let path = ray
      .shortest_path(alice.id, bob.id)
      .via("FOLLOWS")
      .expect("expected value")
      .find();

    assert!(!path.found);
    assert!(path.path.is_empty());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_has_path() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    assert!(ray
      .has_path(alice.id, bob.id, Some("FOLLOWS"))
      .expect("expected value"));
    assert!(!ray
      .has_path(alice.id, charlie.id, Some("FOLLOWS"))
      .expect("expected value"));
    assert!(!ray
      .has_path(bob.id, alice.id, Some("FOLLOWS"))
      .expect("expected value")); // No reverse

    ray.close().expect("expected value");
  }

  #[test]
  fn test_reachable_from() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create: alice -> bob -> charlie -> dave
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");
    let dave = ray
      .create_node("User", "dave", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(bob.id, "FOLLOWS", charlie.id)
      .expect("expected value");
    ray
      .link(charlie.id, "FOLLOWS", dave.id)
      .expect("expected value");

    // Reachable within 2 hops
    let reachable = ray
      .reachable_from(alice.id, 2, Some("FOLLOWS"))
      .expect("expected value");
    assert!(reachable.contains(&bob.id));
    assert!(reachable.contains(&charlie.id));
    assert!(!reachable.contains(&dave.id)); // 3 hops away

    // Reachable within 3 hops
    let reachable_3 = ray
      .reachable_from(alice.id, 3, Some("FOLLOWS"))
      .expect("expected value");
    assert!(reachable_3.contains(&dave.id));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_k_shortest_paths() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a diamond: alice -> bob -> dave, alice -> charlie -> dave
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");
    let dave = ray
      .create_node("User", "dave", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", charlie.id)
      .expect("expected value");
    ray
      .link(bob.id, "FOLLOWS", dave.id)
      .expect("expected value");
    ray
      .link(charlie.id, "FOLLOWS", dave.id)
      .expect("expected value");

    // Find 2 shortest paths
    let paths = ray
      .shortest_path(alice.id, dave.id)
      .via("FOLLOWS")
      .expect("expected value")
      .find_k_shortest(2);

    assert_eq!(paths.len(), 2);
    assert!(paths[0].found);
    assert!(paths[1].found);
    // Both paths have same length (2 edges)
    assert_eq!(paths[0].edges.len(), 2);
    assert_eq!(paths[1].edges.len(), 2);

    ray.close().expect("expected value");
  }

  // ============================================================================
  // Batch Operation Tests
  // ============================================================================

  #[test]
  fn test_batch_create_nodes() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create multiple nodes in a batch
    let results = ray
      .batch(vec![
        BatchOp::CreateNode {
          node_type: "User".into(),
          key_suffix: "alice".into(),
          props: HashMap::new(),
        },
        BatchOp::CreateNode {
          node_type: "User".into(),
          key_suffix: "bob".into(),
          props: HashMap::new(),
        },
        BatchOp::CreateNode {
          node_type: "Post".into(),
          key_suffix: "post1".into(),
          props: HashMap::new(),
        },
      ])
      .expect("expected value");

    assert_eq!(results.len(), 3);

    // Verify all nodes were created
    assert_eq!(ray.count_nodes(), 3);
    assert!(ray.get("User", "alice").expect("expected value").is_some());
    assert!(ray.get("User", "bob").expect("expected value").is_some());
    assert!(ray.get("Post", "post1").expect("expected value").is_some());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_create_and_link() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // First batch: create nodes
    let results = ray
      .batch(vec![
        BatchOp::CreateNode {
          node_type: "User".into(),
          key_suffix: "alice".into(),
          props: HashMap::new(),
        },
        BatchOp::CreateNode {
          node_type: "User".into(),
          key_suffix: "bob".into(),
          props: HashMap::new(),
        },
      ])
      .expect("expected value");

    // Extract node IDs from results
    let alice_id = match &results[0] {
      BatchResult::NodeCreated(node_ref) => node_ref.id(),
      _ => panic!("Expected NodeCreated"),
    };
    let bob_id = match &results[1] {
      BatchResult::NodeCreated(node_ref) => node_ref.id(),
      _ => panic!("Expected NodeCreated"),
    };

    // Second batch: create edge
    ray
      .batch(vec![BatchOp::Link {
        src: alice_id,
        edge_type: "FOLLOWS".into(),
        dst: bob_id,
      }])
      .expect("expected value");

    // Verify edge was created
    assert!(ray
      .has_edge(alice_id, "FOLLOWS", bob_id)
      .expect("expected value"));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_link_with_props() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    let mut props = HashMap::new();
    props.insert("weight".to_string(), PropValue::F64(0.9));
    props.insert("since".to_string(), PropValue::String("2025".into()));

    ray
      .batch(vec![BatchOp::LinkWithProps {
        src: alice.id,
        edge_type: "FOLLOWS".into(),
        dst: bob.id,
        props,
      }])
      .expect("expected value");

    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.9)));

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, Some(PropValue::String("2025".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_set_properties() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node
    let user = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");

    // Batch set properties
    ray
      .batch(vec![
        BatchOp::SetProp {
          node_id: user.id,
          prop_name: "name".into(),
          value: PropValue::String("Alice".into()),
        },
        BatchOp::SetProp {
          node_id: user.id,
          prop_name: "age".into(),
          value: PropValue::I64(30),
        },
      ])
      .expect("expected value");

    // Verify properties
    assert_eq!(
      ray.prop(user.id, "name"),
      Some(PropValue::String("Alice".into()))
    );
    assert_eq!(ray.prop(user.id, "age"), Some(PropValue::I64(30)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_set_edge_properties() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    ray
      .batch(vec![
        BatchOp::SetEdgeProp {
          src: alice.id,
          edge_type: "FOLLOWS".into(),
          dst: bob.id,
          prop_name: "weight".into(),
          value: PropValue::F64(0.75),
        },
        BatchOp::SetEdgeProp {
          src: alice.id,
          edge_type: "FOLLOWS".into(),
          dst: bob.id,
          prop_name: "since".into(),
          value: PropValue::String("2024".into()),
        },
      ])
      .expect("expected value");

    assert_eq!(
      ray
        .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
        .expect("expected value"),
      Some(PropValue::F64(0.75))
    );
    assert_eq!(
      ray
        .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
        .expect("expected value"),
      Some(PropValue::String("2024".into()))
    );

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_set_edge_props() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    let mut props = HashMap::new();
    props.insert("weight".to_string(), PropValue::F64(0.33));
    props.insert("since".to_string(), PropValue::String("2023".into()));

    ray
      .batch(vec![BatchOp::SetEdgeProps {
        src: alice.id,
        edge_type: "FOLLOWS".into(),
        dst: bob.id,
        props,
      }])
      .expect("expected value");

    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.33)));

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, Some(PropValue::String("2023".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_mixed_operations() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create initial nodes
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Mixed batch: link, set prop, create node, unlink
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    let results = ray
      .batch(vec![
        BatchOp::SetProp {
          node_id: alice.id,
          prop_name: "name".into(),
          value: PropValue::String("Alice".into()),
        },
        BatchOp::CreateNode {
          node_type: "User".into(),
          key_suffix: "charlie".into(),
          props: HashMap::new(),
        },
        BatchOp::Unlink {
          src: alice.id,
          edge_type: "FOLLOWS".into(),
          dst: bob.id,
        },
      ])
      .expect("expected value");

    assert_eq!(results.len(), 3);

    // Verify results
    assert_eq!(
      ray.prop(alice.id, "name"),
      Some(PropValue::String("Alice".into()))
    );
    assert!(ray
      .get("User", "charlie")
      .expect("expected value")
      .is_some());
    assert!(!ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_batch_delete_operations() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create nodes and edges
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Batch delete
    let results = ray
      .batch(vec![
        BatchOp::Unlink {
          src: alice.id,
          edge_type: "FOLLOWS".into(),
          dst: bob.id,
        },
        BatchOp::DeleteNode { node_id: bob.id },
      ])
      .expect("expected value");

    // Verify
    match &results[0] {
      BatchResult::EdgeRemoved(removed) => assert!(*removed),
      _ => panic!("Expected EdgeRemoved"),
    }
    match &results[1] {
      BatchResult::NodeDeleted(deleted) => assert!(*deleted),
      _ => panic!("Expected NodeDeleted"),
    }

    assert!(!ray.exists(bob.id));

    ray.close().expect("expected value");
  }

  // ============================================================================
  // Transaction Tests
  // ============================================================================

  #[test]
  fn test_transaction_basic() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Execute a transaction
    let (alice, bob) = ray
      .transaction(|ctx| {
        let alice = ctx.create_node("User", "alice", HashMap::new())?;
        let bob = ctx.create_node("User", "bob", HashMap::new())?;
        ctx.link(alice.id, "FOLLOWS", bob.id)?;
        Ok((alice, bob))
      })
      .expect("expected value");

    // Verify results
    assert!(ray.exists(alice.id));
    assert!(ray.exists(bob.id));
    assert!(ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_transaction_with_properties() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create node with properties in transaction
    let alice = ray
      .transaction(|ctx| {
        let mut props = HashMap::new();
        props.insert("name".to_string(), PropValue::String("Alice".into()));
        let alice = ctx.create_node("User", "alice", props)?;
        ctx.set_prop(alice.id, "age", PropValue::I64(30))?;
        Ok(alice)
      })
      .expect("expected value");

    // Verify properties
    assert_eq!(
      ray.prop(alice.id, "name"),
      Some(PropValue::String("Alice".into()))
    );
    assert_eq!(ray.prop(alice.id, "age"), Some(PropValue::I64(30)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_transaction_rollback_on_error() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Transaction that fails partway through
    let result: Result<()> = ray.transaction(|ctx| {
      ctx.create_node("User", "alice", HashMap::new())?;
      // This should fail - unknown node type
      ctx.create_node("UnknownType", "bob", HashMap::new())?;
      Ok(())
    });

    // Transaction should have failed
    assert!(result.is_err());

    // Alice should NOT exist because the transaction was rolled back
    // Note: Due to WAL-based implementation, rollback happens at commit time
    // so we need to verify the final state

    ray.close().expect("expected value");
  }

  #[test]
  fn test_transaction_read_operations() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create some data first
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    ray
      .set_prop(alice.id, "name", PropValue::String("Alice".into()))
      .expect("expected value");

    // Transaction that reads and writes
    let name = ray
      .transaction(|ctx| {
        // Read existing data
        let existing = ctx.get("User", "alice")?;
        assert!(existing.is_some());

        let name = ctx.prop(alice.id, "name")?;
        assert!(ctx.exists(alice.id));

        // Create new node
        ctx.create_node("User", "bob", HashMap::new())?;

        Ok(name)
      })
      .expect("expected value");

    assert_eq!(name, Some(PropValue::String("Alice".into())));
    assert!(ray.get("User", "bob").expect("expected value").is_some());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_transaction_edge_operations() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create nodes first
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");

    // Link edges in transaction
    ray
      .transaction(|ctx| {
        ctx.link(alice.id, "FOLLOWS", bob.id)?;
        ctx.link(bob.id, "FOLLOWS", charlie.id)?;
        Ok(())
      })
      .expect("expected value");

    // Verify edges exist after commit
    assert!(ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));
    assert!(ray
      .has_edge(bob.id, "FOLLOWS", charlie.id)
      .expect("expected value"));
    assert!(!ray
      .has_edge(alice.id, "FOLLOWS", charlie.id)
      .expect("expected value"));

    // Test unlink in transaction
    ray
      .transaction(|ctx| {
        ctx.unlink(alice.id, "FOLLOWS", bob.id)?;
        Ok(())
      })
      .expect("expected value");

    // Verify edge was removed
    assert!(!ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));
    // Other edge still exists
    assert!(ray
      .has_edge(bob.id, "FOLLOWS", charlie.id)
      .expect("expected value"));

    ray.close().expect("expected value");
  }

  // ============================================================================
  // TxBuilder Tests
  // ============================================================================

  #[test]
  fn test_tx_builder() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Use the builder pattern
    let results = ray
      .tx()
      .create_node("User", "alice", HashMap::new())
      .create_node("User", "bob", HashMap::new())
      .execute(&mut ray)
      .expect("expected value");

    assert_eq!(results.len(), 2);

    // Extract IDs and create edges
    let alice_id = match &results[0] {
      BatchResult::NodeCreated(node_ref) => node_ref.id(),
      _ => panic!("Expected NodeCreated"),
    };
    let bob_id = match &results[1] {
      BatchResult::NodeCreated(node_ref) => node_ref.id(),
      _ => panic!("Expected NodeCreated"),
    };

    ray
      .tx()
      .link(alice_id, "FOLLOWS", bob_id)
      .set_prop(alice_id, "name", PropValue::String("Alice".into()))
      .execute(&mut ray)
      .expect("expected value");

    assert!(ray
      .has_edge(alice_id, "FOLLOWS", bob_id)
      .expect("expected value"));
    assert_eq!(
      ray.prop(alice_id, "name"),
      Some(PropValue::String("Alice".into()))
    );

    ray.close().expect("expected value");
  }

  #[test]
  fn test_tx_builder_into_ops() {
    // Test that into_ops returns the operations without executing
    let ops = TxBuilder::default()
      .create_node("User", "alice", HashMap::new())
      .link(1, "FOLLOWS", 2)
      .set_prop(1, "name", PropValue::String("Test".into()))
      .into_ops();

    assert_eq!(ops.len(), 3);
  }

  // ============================================================================
  // Edge Property Tests
  // ============================================================================

  #[test]
  fn test_link_with_props() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Link with properties
    let mut props = HashMap::new();
    props.insert("weight".to_string(), PropValue::F64(0.8));
    props.insert("since".to_string(), PropValue::String("2024".into()));

    ray
      .link_with_props(alice.id, "FOLLOWS", bob.id, props)
      .expect("expected value");

    // Verify edge exists
    assert!(ray
      .has_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value"));

    // Verify edge properties
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.8)));

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, Some(PropValue::String("2024".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_set_edge_prop() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Create edge without properties
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Set edge property
    ray
      .set_edge_prop(alice.id, "FOLLOWS", bob.id, "weight", PropValue::F64(0.5))
      .expect("expected value");

    // Get edge property
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.5)));

    // Update edge property
    ray
      .set_edge_prop(alice.id, "FOLLOWS", bob.id, "weight", PropValue::F64(0.9))
      .expect("expected value");
    let new_weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(new_weight, Some(PropValue::F64(0.9)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_set_edge_props() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    let mut props = HashMap::new();
    props.insert("weight".to_string(), PropValue::F64(0.6));
    props.insert("since".to_string(), PropValue::String("2024".into()));
    ray
      .set_edge_props(alice.id, "FOLLOWS", bob.id, props)
      .expect("expected value");

    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.6)));

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, Some(PropValue::String("2024".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_edge_props() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Create edge with properties
    let mut props = HashMap::new();
    props.insert("weight".to_string(), PropValue::F64(0.7));
    props.insert("type".to_string(), PropValue::String("friend".into()));
    ray
      .link_with_props(alice.id, "FOLLOWS", bob.id, props)
      .expect("expected value");

    // Get all properties
    let all_props = ray
      .edge_props(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    assert!(all_props.is_some());

    let all_props = all_props.expect("expected value");
    assert_eq!(all_props.get("weight"), Some(&PropValue::F64(0.7)));
    assert_eq!(
      all_props.get("type"),
      Some(&PropValue::String("friend".into()))
    );

    ray.close().expect("expected value");
  }

  #[test]
  fn test_del_edge_prop() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Create edge with property
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .set_edge_prop(alice.id, "FOLLOWS", bob.id, "weight", PropValue::F64(0.5))
      .expect("expected value");

    // Verify property exists
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.5)));

    // Delete property
    ray
      .del_edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");

    // Verify property is gone
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, None);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_edge_prop_nonexistent_edge() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Try to get prop on nonexistent edge - should fail gracefully
    // First we need to create the prop key
    ray
      .set_edge_prop(alice.id, "FOLLOWS", bob.id, "weight", PropValue::F64(0.5))
      .ok();

    // Edge doesn't exist, so getting props should return None
    let _props = ray
      .edge_props(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    // The edge was implicitly created when we set the prop, so it exists now
    // Let's test with a truly nonexistent edge
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");
    let props2 = ray
      .edge_props(alice.id, "FOLLOWS", charlie.id)
      .expect("expected value");
    assert!(props2.is_none());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_edge_builder() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Create edge first
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Update edge properties using the builder
    ray
      .update_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .set("weight", PropValue::F64(0.9))
      .set("since", PropValue::String("2024".into()))
      .execute()
      .expect("expected value");

    // Verify properties were set
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.9)));

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, Some(PropValue::String("2024".into())));

    // Update with unset
    ray
      .update_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .set("weight", PropValue::F64(0.5))
      .unset("since")
      .execute()
      .expect("expected value");

    // Verify update and unset
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.5)));

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, None);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_edge_builder_set_all() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Create edge
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Update using set_all
    let mut props = HashMap::new();
    props.insert("weight".to_string(), PropValue::F64(0.8));
    props.insert("type".to_string(), PropValue::String("close_friend".into()));

    ray
      .update_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .set_all(props)
      .execute()
      .expect("expected value");

    // Verify
    let all_props = ray
      .edge_props(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .expect("expected value");
    assert_eq!(all_props.get("weight"), Some(&PropValue::F64(0.8)));
    assert_eq!(
      all_props.get("type"),
      Some(&PropValue::String("close_friend".into()))
    );

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_edge_builder_empty() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    // Create edge
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Empty update should succeed (no-op)
    ray
      .update_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .execute()
      .expect("expected value");

    ray.close().expect("expected value");
  }

  #[test]
  fn test_upsert_edge_builder() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");

    ray
      .upsert_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .set("since", PropValue::I64(2020))
      .execute()
      .expect("expected value");

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, Some(PropValue::I64(2020)));

    ray
      .upsert_edge(alice.id, "FOLLOWS", bob.id)
      .expect("expected value")
      .set("weight", PropValue::F64(0.5))
      .unset("since")
      .execute()
      .expect("expected value");

    let since = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "since")
      .expect("expected value");
    assert_eq!(since, None);
    let weight = ray
      .edge_prop(alice.id, "FOLLOWS", bob.id, "weight")
      .expect("expected value");
    assert_eq!(weight, Some(PropValue::F64(0.5)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_insert_builder_returning() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Insert with returning
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Alice".into()));
    props.insert("age".to_string(), PropValue::I64(30));

    let alice = ray
      .insert("User")
      .expect("expected value")
      .values("alice", props)
      .expect("expected value")
      .returning()
      .expect("expected value");

    // Verify the returned node
    assert!(alice.id > 0);
    assert_eq!(alice.key, Some("user:alice".to_string()));
    assert_eq!(alice.node_type.as_ref(), "User");

    // Verify properties were set
    let name = ray.prop(alice.id, "name");
    assert_eq!(name, Some(PropValue::String("Alice".into())));

    let age = ray.prop(alice.id, "age");
    assert_eq!(age, Some(PropValue::I64(30)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_insert_builder_execute() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Insert without returning
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Bob".into()));

    ray
      .insert("User")
      .expect("expected value")
      .values("bob", props)
      .expect("expected value")
      .execute()
      .expect("expected value");

    // Verify node was created
    let bob = ray.get("User", "bob").expect("expected value");
    assert!(bob.is_some());

    let bob = bob.expect("expected value");
    let name = ray.prop(bob.id, "name");
    assert_eq!(name, Some(PropValue::String("Bob".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_insert_builder_values_many() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Insert multiple nodes
    let mut alice_props = HashMap::new();
    alice_props.insert("name".to_string(), PropValue::String("Alice".into()));

    let mut bob_props = HashMap::new();
    bob_props.insert("name".to_string(), PropValue::String("Bob".into()));

    let mut charlie_props = HashMap::new();
    charlie_props.insert("name".to_string(), PropValue::String("Charlie".into()));

    let users = ray
      .insert("User")
      .expect("expected value")
      .values_many(vec![
        ("alice", alice_props),
        ("bob", bob_props),
        ("charlie", charlie_props),
      ])
      .expect("expected value")
      .returning()
      .expect("expected value");

    // Verify all nodes were created
    assert_eq!(users.len(), 3);
    assert_eq!(users[0].key, Some("user:alice".to_string()));
    assert_eq!(users[1].key, Some("user:bob".to_string()));
    assert_eq!(users[2].key, Some("user:charlie".to_string()));

    // Verify count
    assert_eq!(ray.count_nodes(), 3);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_insert_builder_empty_values_many() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Empty insert should succeed
    let users = ray
      .insert("User")
      .expect("expected value")
      .values_many(vec![])
      .expect("expected value")
      .returning()
      .expect("expected value");

    assert_eq!(users.len(), 0);
    assert_eq!(ray.count_nodes(), 0);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_check_empty_database() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    let result = ray.check().expect("expected value");
    assert!(result.valid);
    assert!(result.errors.is_empty());
    // Should warn about an empty database
    assert!(result
      .warnings
      .iter()
      .any(|w| w.contains("No nodes in database")));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_check_valid_database() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create some nodes and edges
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    let charlie = ray
      .create_node("User", "charlie", HashMap::new())
      .expect("expected value");

    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");
    ray
      .link(bob.id, "FOLLOWS", charlie.id)
      .expect("expected value");
    ray
      .link(charlie.id, "FOLLOWS", alice.id)
      .expect("expected value");

    // Check should pass
    let result = ray.check().expect("expected value");
    assert!(
      result.valid,
      "Expected valid database, got errors: {:?}",
      result.errors
    );
    assert!(result.errors.is_empty());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_check_with_properties() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create nodes with properties
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Alice".into()));
    props.insert("age".to_string(), PropValue::I64(30));
    let alice = ray
      .create_node("User", "alice", props)
      .expect("expected value");

    let mut props2 = HashMap::new();
    props2.insert("name".to_string(), PropValue::String("Bob".into()));
    let bob = ray
      .create_node("User", "bob", props2)
      .expect("expected value");

    // Create edge with properties
    let mut edge_props = HashMap::new();
    edge_props.insert("weight".to_string(), PropValue::F64(0.9));
    ray
      .link_with_props(alice.id, "FOLLOWS", bob.id, edge_props)
      .expect("expected value");

    // Check should pass
    let result = ray.check().expect("expected value");
    assert!(
      result.valid,
      "Expected valid database, got errors: {:?}",
      result.errors
    );
    assert!(result.errors.is_empty());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_by_ref() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Alice".into()));
    props.insert("age".to_string(), PropValue::I64(30));
    let alice = ray
      .create_node("User", "alice", props)
      .expect("expected value");

    // Update by reference
    ray
      .update(&alice)
      .expect("expected value")
      .set("name", PropValue::String("Alice Updated".into()))
      .set("age", PropValue::I64(31))
      .execute()
      .expect("expected value");

    // Verify updates
    let name = ray.prop(alice.id, "name");
    assert_eq!(name, Some(PropValue::String("Alice Updated".into())));

    let age = ray.prop(alice.id, "age");
    assert_eq!(age, Some(PropValue::I64(31)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_by_key() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Bob".into()));
    ray
      .create_node("User", "bob", props)
      .expect("expected value");

    // Update by key
    ray
      .update_by_key("User", "bob")
      .expect("expected value")
      .set("name", PropValue::String("Bob Updated".into()))
      .set("age", PropValue::I64(25))
      .execute()
      .expect("expected value");

    // Verify updates
    let bob = ray
      .get("User", "bob")
      .expect("expected value")
      .expect("expected value");
    let name = ray.prop(bob.id, "name");
    assert_eq!(name, Some(PropValue::String("Bob Updated".into())));

    let age = ray.prop(bob.id, "age");
    assert_eq!(age, Some(PropValue::I64(25)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_by_id() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Charlie".into()));
    let charlie = ray
      .create_node("User", "charlie", props)
      .expect("expected value");

    // Update by ID
    ray
      .update_by_id(charlie.id)
      .expect("expected value")
      .set("name", PropValue::String("Charlie Updated".into()))
      .execute()
      .expect("expected value");

    // Verify updates
    let name = ray.prop(charlie.id, "name");
    assert_eq!(name, Some(PropValue::String("Charlie Updated".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_upsert_node_by_id_builder() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create by ID
    ray
      .upsert_by_id("User", 42)
      .expect("expected value")
      .set("name", PropValue::String("Alice".into()))
      .set("age", PropValue::I64(30))
      .execute()
      .expect("expected value");

    assert!(ray.exists(42));

    let name = ray.prop(42, "name");
    assert_eq!(name, Some(PropValue::String("Alice".into())));
    let age = ray.prop(42, "age");
    assert_eq!(age, Some(PropValue::I64(30)));

    // Update same ID
    ray
      .upsert_by_id("User", 42)
      .expect("expected value")
      .set("age", PropValue::I64(31))
      .unset("name")
      .execute()
      .expect("expected value");

    let name = ray.prop(42, "name");
    assert_eq!(name, None);
    let age = ray.prop(42, "age");
    assert_eq!(age, Some(PropValue::I64(31)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_unset() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node with properties
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Dave".into()));
    props.insert("age".to_string(), PropValue::I64(40));
    let dave = ray
      .create_node("User", "dave", props)
      .expect("expected value");

    // Verify properties exist
    assert!(ray.prop(dave.id, "age").is_some());

    // Update with unset
    ray
      .update(&dave)
      .expect("expected value")
      .set("name", PropValue::String("Dave Updated".into()))
      .unset("age")
      .execute()
      .expect("expected value");

    // Verify name updated and age removed
    let name = ray.prop(dave.id, "name");
    assert_eq!(name, Some(PropValue::String("Dave Updated".into())));

    let age = ray.prop(dave.id, "age");
    assert_eq!(age, None);

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_set_all() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node
    let eve = ray
      .create_node("User", "eve", HashMap::new())
      .expect("expected value");

    // Update with set_all
    let mut updates = HashMap::new();
    updates.insert("name".to_string(), PropValue::String("Eve".into()));
    updates.insert("age".to_string(), PropValue::I64(28));

    ray
      .update(&eve)
      .expect("expected value")
      .set_all(updates)
      .execute()
      .expect("expected value");

    // Verify all properties set
    let name = ray.prop(eve.id, "name");
    assert_eq!(name, Some(PropValue::String("Eve".into())));

    let age = ray.prop(eve.id, "age");
    assert_eq!(age, Some(PropValue::I64(28)));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_nonexistent() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Try to update non-existent node by ID
    let result = ray.update_by_id(999999);
    assert!(result.is_err());

    // Try to update non-existent node by key
    let result = ray.update_by_key("User", "nonexistent");
    assert!(result.is_err());

    ray.close().expect("expected value");
  }

  #[test]
  fn test_update_node_empty() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create a node
    let mut props = HashMap::new();
    props.insert("name".to_string(), PropValue::String("Frank".into()));
    let frank = ray
      .create_node("User", "frank", props)
      .expect("expected value");

    // Empty update should succeed (no-op)
    ray
      .update(&frank)
      .expect("expected value")
      .execute()
      .expect("expected value");

    // Verify nothing changed
    let name = ray.prop(frank.id, "name");
    assert_eq!(name, Some(PropValue::String("Frank".into())));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_describe() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create some data
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Get description
    let desc = ray.describe();

    // Should contain path
    assert!(desc.contains("KiteDB at"));
    // Should mention format
    assert!(desc.contains("format"));
    // Should list node types
    assert!(desc.contains("User"));
    // Should list edge types
    assert!(desc.contains("FOLLOWS"));
    // Should include stats
    assert!(desc.contains("Nodes:"));
    assert!(desc.contains("Edges:"));

    ray.close().expect("expected value");
  }

  #[test]
  fn test_stats() {
    let temp_dir = tempdir().expect("expected value");
    let options = create_test_schema();

    let mut ray = Kite::open(temp_db_path(&temp_dir), options).expect("expected value");

    // Create some data
    let alice = ray
      .create_node("User", "alice", HashMap::new())
      .expect("expected value");
    let bob = ray
      .create_node("User", "bob", HashMap::new())
      .expect("expected value");
    ray
      .link(alice.id, "FOLLOWS", bob.id)
      .expect("expected value");

    // Get stats
    let stats = ray.stats();

    // Should report correct counts (snapshot + delta)
    assert!(stats.snapshot_nodes + stats.delta_nodes_created as u64 >= 2);
    assert!(stats.snapshot_edges + stats.delta_edges_added as u64 >= 1);

    ray.close().expect("expected value");
  }
}