rocksolid 3.0.0

use crate::batch::BatchWriter;
use crate::bytes::AsBytes;
use crate::config::{RocksDbCFStoreConfig, convert_recovery_mode, default_full_merge, default_partial_merge};
use crate::deserialize_kv_expiry;
use crate::error::{StoreError, StoreResult};
use crate::iter::helpers::{GeneralFactory, IterationHelper, PrefixFactory};
use crate::iter::seekable::SeekableRows;
use crate::iter::{IterConfig, IterationResult};
use crate::serialization::{deserialize_kv, deserialize_value, serialize_key, serialize_value};
use crate::tuner::{PatternTuner, Tunable};
use crate::types::{IterationControlDecision, MergeValue, ValueWithExpiry};

use bytevec::ByteDecodable;
use rocksdb::{ColumnFamilyDescriptor, DB, Direction, Options as RocksDbOptions, ReadOptions, WriteBatch};
use serde::{Serialize, de::DeserializeOwned};
use std::collections::HashSet;
use std::hash::Hash;
use std::{collections::HashMap, fmt::Debug, path::Path, sync::Arc};

// --- CfOperations Trait (Public API for CF-aware operations) ---

pub trait CFOperations {
  // --- Read Operations ---
  fn get<K, V>(&self, cf_name: &str, key: K) -> StoreResult<Option<V>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: DeserializeOwned + Debug;

  fn get_raw<K>(&self, cf_name: &str, key: K) -> StoreResult<Option<Vec<u8>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  fn get_with_expiry<K, V>(&self, cf_name: &str, key: K) -> StoreResult<Option<ValueWithExpiry<V>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + DeserializeOwned + Debug;

  fn exists<K>(&self, cf_name: &str, key: K) -> StoreResult<bool>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  // --- Multi Get Operations ---
  fn multiget<K, V>(&self, cf_name: &str, keys: &[K]) -> StoreResult<Vec<Option<V>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug + Clone, // Clone for processing keys with results
    V: DeserializeOwned + Debug;

  fn multiget_raw<K>(&self, cf_name: &str, keys: &[K]) -> StoreResult<Vec<Option<Vec<u8>>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  fn multiget_with_expiry<K, V>(&self, cf_name: &str, keys: &[K]) -> StoreResult<Vec<Option<ValueWithExpiry<V>>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug + Clone,
    V: Serialize + DeserializeOwned + Debug;

  // --- Write Operations ---
  fn put<K, V>(&self, cf_name: &str, key: K, value: &V) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + Debug;

  fn put_raw<K>(&self, cf_name: &str, key: K, raw_value: &[u8]) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  fn put_with_expiry<K, V>(&self, cf_name: &str, key: K, value: &V, expire_time: u64) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + DeserializeOwned + Debug;

  fn delete<K>(&self, cf_name: &str, key: K) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  fn delete_range<K>(&self, cf_name: &str, start_key: K, end_key: K) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  fn merge<K, PatchVal>(&self, cf_name: &str, key: K, merge_value: &MergeValue<PatchVal>) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    PatchVal: Serialize + Debug;

  fn merge_raw<K>(&self, cf_name: &str, key: K, raw_merge_operand: &[u8]) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug;

  fn merge_with_expiry<K, V>(&self, cf_name: &str, key: K, value: &V, expire_time: u64) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + DeserializeOwned + Debug;

  // --- Iterator / Find Operations ---

  /// General purpose iteration method.
  ///
  /// The behavior and output type depend on `config.mode`.
  /// - `IterationMode::Deserialize`: Returns `IterationResult::DeserializedItems`.
  /// - `IterationMode::Raw`: Returns `IterationResult::RawItems`.
  /// - `IterationMode::ControlOnly`: Returns `IterationResult::EffectCompleted`.
  fn iterate<'store_lt, SerKey, OutK, OutV>(
    &'store_lt self,
    config: IterConfig<'store_lt, SerKey, OutK, OutV>,
  ) -> Result<IterationResult<'store_lt, OutK, OutV>, StoreError>
  where
    SerKey: AsBytes + Hash + Eq + PartialEq + Debug,
    OutK: DeserializeOwned + Debug + 'store_lt,
    OutV: DeserializeOwned + Debug + 'store_lt;

  fn find_by_prefix<Key, Val>(&self, cf_name: &str, prefix: &Key, direction: Direction) -> StoreResult<Vec<(Key, Val)>>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug + Clone,
    Val: DeserializeOwned + Debug;

  fn find_from<Key, Val, ControlFn>(
    &self,
    cf_name: &str,
    start_key: Key,
    direction: Direction,
    control_fn: ControlFn,
  ) -> StoreResult<Vec<(Key, Val)>>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug,
    Val: DeserializeOwned + Debug,
    ControlFn: FnMut(&[u8], &[u8], usize) -> IterationControlDecision + 'static;

  fn find_from_with_expire_val<Key, Val, ControlFn>(
    &self,
    cf_name: &str,
    start: &Key,
    reverse: bool,
    control_fn: ControlFn,
  ) -> Result<Vec<(Key, ValueWithExpiry<Val>)>, String>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug + Clone,
    Val: DeserializeOwned + Debug,
    ControlFn: FnMut(&[u8], &[u8], usize) -> IterationControlDecision + 'static;

  fn find_by_prefix_with_expire_val<Key, Val, ControlFn>(
    &self,
    cf_name: &str,
    start: &Key,
    reverse: bool,
    control_fn: ControlFn,
  ) -> Result<Vec<(Key, ValueWithExpiry<Val>)>, String>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug + Clone,
    Val: DeserializeOwned + Debug,
    ControlFn: FnMut(&[u8], &[u8], usize) -> IterationControlDecision + 'static;
}

/// The foundational, public, Column Family (CF)-aware key-value store.
pub struct RocksDbCFStore {
  db: Arc<DB>,
  cf_names: HashSet<String>,
  path: String,
}

impl Debug for RocksDbCFStore {
  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    f.debug_struct("RocksDbCFStore")
      .field("path", &self.path)
      .field("db", &"<Arc<rocksdb::DB>>")
      .field("cf_names", &self.cf_names.iter().collect::<Vec<&String>>())
      .finish()
  }
}

impl RocksDbCFStore {
  /// Opens or creates a RocksDB database with the specified Column Families and configurations.
  ///
  /// # Arguments
  /// * `cfg` - The configuration for the CF-aware store.
  ///
  /// # Errors
  /// Returns `StoreError` if opening fails, CF configuration is invalid, or CFs are not found.
  pub fn open(cfg: RocksDbCFStoreConfig) -> StoreResult<Self> {
    log::info!(
      "Opening RocksDbCFStore at path: '{}'. CFs to open: {:?}",
      cfg.path,
      cfg.column_families_to_open
    );

    let mut db_opts_tunable = Tunable::new(RocksDbOptions::default());
    db_opts_tunable.inner.create_if_missing(cfg.create_if_missing);
    db_opts_tunable
      .inner
      .create_missing_column_families(cfg.create_if_missing);

    if let Some(p) = cfg.parallelism {
      db_opts_tunable.set_increase_parallelism(p);
    }
    if let Some(mode) = cfg.recovery_mode {
      db_opts_tunable.inner.set_wal_recovery_mode(convert_recovery_mode(mode));
    }
    if let Some(enable_stats) = cfg.enable_statistics {
      if enable_stats {
        db_opts_tunable.inner.enable_statistics();
      } else {
        // There isn't a direct `disable_statistics`. If a profile enables it,
        // and this hard setting is false, the user must ensure the profile
        // doesn't re-enable it or use custom_options to override.
        log::debug!(
          "Hard setting 'enable_statistics: false' noted. Ensure profiles or custom_options respect this if needed."
        );
      }
    }

    if let Some(profile) = &cfg.db_tuning_profile {
      profile.tune_db_opts(&cfg.path, &mut db_opts_tunable);
    }

    let mut cf_options_map_tunable: HashMap<String, Tunable<RocksDbOptions>> = HashMap::new();

    let cfs_to_actually_open = cfg.column_families_to_open.clone();

    for cf_name_str in &cfs_to_actually_open {
      let mut current_cf_tunable = Tunable::new(RocksDbOptions::default());
      let cf_config_for_this_cf = cfg.column_family_configs.get(cf_name_str);

      let effective_profile = cf_config_for_this_cf
        .and_then(|c| c.tuning_profile.as_ref())
        .or_else(|| cfg.db_tuning_profile.as_ref());

      if let Some(profile) = effective_profile {
        profile.tune_cf_opts(cf_name_str, &mut current_cf_tunable);
      }
      cf_options_map_tunable.insert(cf_name_str.clone(), current_cf_tunable);
    }

    if cfs_to_actually_open.contains(&rocksdb::DEFAULT_COLUMN_FAMILY_NAME.to_string())
      && !cf_options_map_tunable.contains_key(rocksdb::DEFAULT_COLUMN_FAMILY_NAME)
    {
      let mut default_cf_tunable = Tunable::new(RocksDbOptions::default());
      if let Some(profile) = &cfg.db_tuning_profile {
        profile.tune_cf_opts(rocksdb::DEFAULT_COLUMN_FAMILY_NAME, &mut default_cf_tunable);
      }
      cf_options_map_tunable.insert(rocksdb::DEFAULT_COLUMN_FAMILY_NAME.to_string(), default_cf_tunable);
    }

    if let Some(custom_fn) = &cfg.custom_options_db_and_cf {
      custom_fn(&mut db_opts_tunable, &mut cf_options_map_tunable);
    }

    let raw_db_opts = db_opts_tunable.into_inner();
    let mut raw_cf_options_map: HashMap<String, RocksDbOptions> = cf_options_map_tunable
      .into_iter()
      .map(|(name, tunable_opts)| (name, tunable_opts.into_inner()))
      .collect();
    for (cf_name, cf_specific_config) in &cfg.column_family_configs {
      if let Some(opts_to_modify) = raw_cf_options_map.get_mut(cf_name) {
        if let Some(merge_op_config) = &cf_specific_config.merge_operator {
          opts_to_modify.set_merge_operator(
            &merge_op_config.name,
            merge_op_config.full_merge_fn.unwrap_or(default_full_merge),
            merge_op_config.partial_merge_fn.unwrap_or(default_partial_merge),
          );
          log::debug!("Applied merge operator '{}' to CF '{}'", merge_op_config.name, cf_name);
        }

        if let Some(comparator_choice) = &cf_specific_config.comparator {
          comparator_choice.apply_to_opts(cf_name, opts_to_modify);
        } else {
          log::debug!(
            "No explicit comparator specified for CF '{}'. Using RocksDB default or prior setting.",
            cf_name
          );
        }

        if let Some(filter_router_config) = &cf_specific_config.compaction_filter_router {
          let actual_router_fn_ptr = filter_router_config.filter_fn_ptr;

          let boxed_router_callback = Box::new(
            move |level: u32, key: &[u8], value: &[u8]| -> rocksdb::compaction_filter::Decision {
              actual_router_fn_ptr(level, key, value)
            },
          );

          opts_to_modify.set_compaction_filter(&filter_router_config.name, boxed_router_callback);
          log::debug!(
            "Applied compaction filter router named '{}' to CF '{}'",
            filter_router_config.name,
            cf_name
          );
        }
      }
    }

    let cf_descriptors: Vec<ColumnFamilyDescriptor> = cfs_to_actually_open
          .iter()
          .map(|name_str| {
              let cf_opts = raw_cf_options_map.remove(name_str)
                              .unwrap_or_else(|| {
                                  log::warn!("Options for CF '{}' not found in map, using default. This indicates a potential issue in config processing.", name_str);
                                  RocksDbOptions::default()
                              });
              ColumnFamilyDescriptor::new(name_str, cf_opts)
          })
          .collect();

    if cf_descriptors.is_empty() && cfs_to_actually_open.is_empty() {
      // With an empty descriptor list, `DB::open_cf_descriptors` opens only the default CF
      // with default options; `raw_db_opts` still applies to the DB itself.
      log::info!(
        "Opening DB with CF descriptors. DB options applied. CF descriptors count: {}",
        cf_descriptors.len()
      );
    }

    let db_instance =
      DB::open_cf_descriptors(&raw_db_opts, Path::new(&cfg.path), cf_descriptors).map_err(StoreError::RocksDb)?;

    let db_arc = Arc::new(db_instance);

    let mut cf_handles_map = HashSet::new();
    for cf_name_str in &cfs_to_actually_open {
      cf_handles_map.insert(cf_name_str.clone());
    }

    log::info!("RocksDbCFStore opened successfully at path '{}'", cfg.path);
    Ok(Self {
      db: db_arc,
      cf_names: cf_handles_map,
      path: cfg.path.clone(),
    })
  }

  /// Returns the filesystem path of the database directory.
  pub fn path(&self) -> &str {
    &self.path
  }

  /// Internal helper to get a `ColumnFamily` handle.
  /// Returns `StoreError::UnknownCf` if the handle is not found (i.e., CF was not opened).
  pub fn get_cf_handle(&'_ self, cf_name: &str) -> StoreResult<Arc<rocksdb::BoundColumnFamily<'_>>> {
    return self
      .db
      .cf_handle(cf_name)
      .ok_or_else(|| StoreError::UnknownCf(cf_name.to_string()));
  }

  /// Returns a thread-safe reference (`Arc`) to the underlying `rocksdb::DB` instance.
  /// Useful for operations not directly exposed by `RocksDbCFStore` or for advanced features
  /// like checkpoints, manual compactions, etc.
  pub fn db_raw(&self) -> Arc<DB> {
    self.db.clone()
  }

  /// Flushes the Write-Ahead Log to the filesystem. If `sync` is true, the file is fsynced.
  pub fn flush_wal(&self, sync: bool) -> StoreResult<()> {
    self.db.flush_wal(sync).map_err(StoreError::RocksDb)
  }

  /// Flushes the memtable of a single Column Family to SST files on disk.
  pub fn flush_cf(&self, cf_name: &str) -> StoreResult<()> {
    let handle = self.get_cf_handle(cf_name)?;
    self.db.flush_cf(&handle).map_err(StoreError::RocksDb)
  }

  /// Flushes the memtables of all opened Column Families to SST files on disk.
  pub fn flush_all_cfs(&self) -> StoreResult<()> {
    for cf_name in &self.cf_names {
      self.flush_cf(cf_name)?;
    }
    Ok(())
  }

  /// Creates a `BatchWriter` for this store, targeting a single Column Family.
  pub fn batch_writer(&self, cf_name: &str) -> BatchWriter<'_> {
    BatchWriter::new(self, cf_name.to_string())
  }

  /// Creates a [`MultiCfBatchWriter`](crate::batch::MultiCfBatchWriter) for this store,
  /// allowing atomic writes spanning multiple Column Families in one batch.
  pub fn batch_writer_multi_cf(&self) -> crate::batch::MultiCfBatchWriter<'_> {
    crate::batch::MultiCfBatchWriter::new(self)
  }

  /// Destroys the database files at the given path. Use with extreme caution.
  ///
  /// This method constructs minimal `RocksDbOptions` required for destruction based on the provided config.
  /// Ensure the `RocksDbCFStore` instance is dropped and no other processes are using the DB.
  ///
  /// # Arguments
  /// * `path` - Path to the database directory.
  /// * `cfg` - Configuration for the store, used to derive necessary DB options for destruction.
  ///           Only DB-wide settings from `cfg` (like hard settings, db_profile, custom_options_db part)
  ///           are relevant here, as CF options aren't needed for `DB::destroy`.
  pub fn destroy(path: &Path, cfg: RocksDbCFStoreConfig) -> StoreResult<()> {
    log::warn!("Destroying RocksDB database at path: {}", path.display());

    let mut opts_tunable = Tunable::new(RocksDbOptions::default());

    if let Some(p) = cfg.parallelism {
      opts_tunable.set_increase_parallelism(p);
    }
    if let Some(mode) = cfg.recovery_mode {
      opts_tunable.inner.set_wal_recovery_mode(convert_recovery_mode(mode));
    }
    if let Some(enable_stats) = cfg.enable_statistics {
      if enable_stats {
        opts_tunable.inner.enable_statistics();
      }
    }

    if let Some(profile) = &cfg.db_tuning_profile {
      profile.tune_db_opts(path.to_str().unwrap_or("db_for_destroy"), &mut opts_tunable);
    }

    if let Some(custom_fn) = &cfg.custom_options_db_and_cf {
      let mut empty_cf_opts_map = HashMap::new();
      custom_fn(&mut opts_tunable, &mut empty_cf_opts_map);
    }

    let final_opts = opts_tunable.into_inner();
    DB::destroy(&final_opts, path).map_err(StoreError::RocksDb)?;
    log::info!("Successfully destroyed RocksDB database at path: {}", path.display());
    Ok(())
  }
}

impl CFOperations for RocksDbCFStore {
  // --- Read Operations ---
  fn get<K, V>(&self, cf_name: &str, key: K) -> StoreResult<Option<V>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: DeserializeOwned + Debug,
  {
    let ser_key = serialize_key(key)?;
    let opt_bytes = if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self.db.get_pinned(&ser_key)?
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self.db.get_pinned_cf(&handle, &ser_key)?
    };

    opt_bytes.map_or(Ok(None), |val_bytes| deserialize_value(&val_bytes).map(Some))
  }

  fn get_raw<K>(&self, cf_name: &str, key: K) -> StoreResult<Option<Vec<u8>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    let ser_key = serialize_key(key)?;
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self
        .db
        .get_pinned(&ser_key)
        .map(|opt| opt.map(|slice| slice.to_vec()))
        .map_err(StoreError::RocksDb)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self
        .db
        .get_pinned_cf(&handle, &ser_key)
        .map(|opt| opt.map(|slice| slice.to_vec()))
        .map_err(StoreError::RocksDb)
    }
  }

  fn get_with_expiry<K, V>(&self, cf_name: &str, key: K) -> StoreResult<Option<ValueWithExpiry<V>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + DeserializeOwned + Debug,
  {
    let opt_bytes = self.get_raw(cf_name, key)?;
    opt_bytes.map_or(Ok(None), |bytes| ValueWithExpiry::from_slice(&bytes).map(Some))
  }

  fn exists<K>(&self, cf_name: &str, key: K) -> StoreResult<bool>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    // More efficient to use get_pinned and check for Some presence than key_may_exist
    let ser_key = serialize_key(key)?;
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self
        .db
        .get_pinned(&ser_key)
        .map(|opt| opt.is_some())
        .map_err(StoreError::RocksDb)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self
        .db
        .get_pinned_cf(&handle, &ser_key)
        .map(|opt| opt.is_some())
        .map_err(StoreError::RocksDb)
    }
  }

  // --- Multi Get Operations ---
  fn multiget<K, V>(&self, cf_name: &str, keys: &[K]) -> StoreResult<Vec<Option<V>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug + Clone,
    V: DeserializeOwned + Debug,
  {
    if keys.is_empty() {
      return Ok(Vec::new());
    }
    let serialized_keys_refs: Vec<_> = keys.iter().map(|k| serialize_key(k)).collect::<StoreResult<_>>()?;

    // Default CF uses DB::multi_get; named CFs use multi_get_cf with a per-key handle.
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      let results_from_db = self.db.multi_get(&serialized_keys_refs);
      results_from_db
        .into_iter()
        .map(|opt_db_val| {
          opt_db_val.map_or(Ok(None), |db_val_res| {
            db_val_res.map_or(Ok(None), |opt_vec| {
              deserialize_value(&opt_vec).map(Some)
            })
          })
        })
        .collect()
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      let keys_with_cf: Vec<(&Arc<rocksdb::BoundColumnFamily>, &[u8])> = serialized_keys_refs
        .iter()
        .map(|sk_ref| (&handle, sk_ref.as_slice()))
        .collect();

      let results_from_db = self.db.multi_get_cf_opt(keys_with_cf, &ReadOptions::default());
      results_from_db
        .into_iter()
        .map(|opt_db_val| {
          opt_db_val.map_or(Ok(None), |db_val_res| {
            db_val_res.map_or(Ok(None), |opt_vec| deserialize_value(&opt_vec).map(Some))
          })
        })
        .collect()
    }
  }

  fn multiget_raw<K>(&self, cf_name: &str, keys: &[K]) -> StoreResult<Vec<Option<Vec<u8>>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    if keys.is_empty() {
      return Ok(Vec::new());
    }
    let serialized_keys_refs: Vec<_> = keys.iter().map(|k| serialize_key(k)).collect::<StoreResult<_>>()?;

    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      let results = self.db.multi_get(serialized_keys_refs);
      results
        .into_iter()
        .map(|res_opt_dbvec| res_opt_dbvec.map(|opt_dbvec| opt_dbvec.map(|dbvec| dbvec.to_vec())))
        .collect::<Result<Vec<_>, _>>()
        .map_err(StoreError::RocksDb)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      let keys_with_cf: Vec<(&Arc<rocksdb::BoundColumnFamily>, &[u8])> = serialized_keys_refs
        .iter()
        .map(|sk_ref| (&handle, sk_ref.as_slice()))
        .collect();
      let results = self.db.multi_get_cf_opt(keys_with_cf, &ReadOptions::default());
      results
        .into_iter()
        .map(|res_opt_dbvec| res_opt_dbvec.map(|opt_dbvec| opt_dbvec.map(|dbvec| dbvec.to_vec())))
        .collect::<Result<Vec<_>, _>>()
        .map_err(StoreError::RocksDb)
    }
  }

  fn multiget_with_expiry<K, V>(&self, cf_name: &str, keys: &[K]) -> StoreResult<Vec<Option<ValueWithExpiry<V>>>>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug + Clone,
    V: Serialize + DeserializeOwned + Debug,
  {
    let raw_results = self.multiget_raw(cf_name, keys)?;
    raw_results
      .into_iter()
      .map(|opt_bytes| opt_bytes.map_or(Ok(None), |bytes| ValueWithExpiry::from_slice(&bytes).map(Some)))
      .collect()
  }

  // --- Write Operations ---
  fn put<K, V>(&self, cf_name: &str, key: K, value: &V) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + Debug,
  {
    let ser_key = serialize_key(key)?;
    let ser_val = serialize_value(value)?;
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self.db.put(&ser_key, &ser_val)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self.db.put_cf(&handle, &ser_key, &ser_val)
    }
    .map_err(StoreError::RocksDb)
  }

  fn put_raw<K>(&self, cf_name: &str, key: K, raw_value: &[u8]) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    let ser_key = serialize_key(key)?;
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self.db.put(&ser_key, raw_value)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self.db.put_cf(&handle, &ser_key, raw_value)
    }
    .map_err(StoreError::RocksDb)
  }

  fn put_with_expiry<K, V>(&self, cf_name: &str, key: K, value: &V, expire_time: u64) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + DeserializeOwned + Debug,
  {
    let vwe = ValueWithExpiry::from_value(expire_time, value)?;
    self.put_raw(cf_name, key, &vwe.serialize_for_storage())
  }

  fn delete<K>(&self, cf_name: &str, key: K) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    let ser_key = serialize_key(key)?;
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self.db.delete(&ser_key)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self.db.delete_cf(&handle, &ser_key)
    }
    .map_err(StoreError::RocksDb)
  }

  fn delete_range<K>(&self, cf_name: &str, start_key: K, end_key: K) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    let sk_start = serialize_key(start_key)?;
    let sk_end = serialize_key(end_key)?;
    // RocksDB delete_range_cf needs WriteOptions. For single op, use batch.
    let mut batch = WriteBatch::default();
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      batch.delete_range(sk_start, sk_end);
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      batch.delete_range_cf(&handle, sk_start, sk_end);
    }
    self.db.write(&batch).map_err(StoreError::RocksDb)
  }

  fn merge<K, PatchVal>(&self, cf_name: &str, key: K, merge_value: &MergeValue<PatchVal>) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    PatchVal: Serialize + Debug,
  {
    let ser_key = serialize_key(&key)?;
    let ser_merge_op = serialize_value(merge_value)?;

    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self.db.merge(&ser_key, &ser_merge_op)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self.db.merge_cf(&handle, &ser_key, &ser_merge_op)
    }
    .map_err(StoreError::RocksDb)
  }

  fn merge_raw<K>(&self, cf_name: &str, key: K, raw_merge_operand: &[u8]) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
  {
    let ser_key = serialize_key(key)?;
    if cf_name == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
      self.db.merge(&ser_key, raw_merge_operand)
    } else {
      let handle = self.get_cf_handle(cf_name)?;
      self.db.merge_cf(&handle, &ser_key, raw_merge_operand)
    }
    .map_err(StoreError::RocksDb)
  }

  fn merge_with_expiry<K, V>(&self, cf_name: &str, key: K, value: &V, expire_time: u64) -> StoreResult<()>
  where
    K: AsBytes + Hash + Eq + PartialEq + Debug,
    V: Serialize + DeserializeOwned + Debug,
  {
    let vwe = ValueWithExpiry::from_value(expire_time, value)?;
    self.merge_raw(cf_name, key, &vwe.serialize_for_storage())
  }

  // --- Iterator / Find Operations ---
  fn iterate<'store_lt, SerKey, OutK, OutV>(
    &'store_lt self,
    config: IterConfig<'store_lt, SerKey, OutK, OutV>,
  ) -> Result<IterationResult<'store_lt, OutK, OutV>, StoreError>
  where
    SerKey: AsBytes + Hash + Eq + PartialEq + Debug,
    OutK: DeserializeOwned + Debug + 'store_lt,
    OutV: DeserializeOwned + Debug + 'store_lt,
  {
    let cf_name_for_general = config.cf_name.clone();
    let cf_name_for_prefix = config.cf_name.clone();

    let general_iterator_factory: GeneralFactory<'store_lt> = Box::new(move |mode| {
      let read_opts = ReadOptions::default();
      let iter: Box<dyn SeekableRows + 'store_lt> =
        if cf_name_for_general == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
          Box::new(self.db.iterator_opt(mode, read_opts))
        } else {
          let handle = self.get_cf_handle(&cf_name_for_general)?;
          Box::new(self.db.iterator_cf_opt(&handle, read_opts, mode))
        };
      Ok(iter)
    });

    let prefix_iterator_factory: PrefixFactory<'store_lt> = Box::new(move |prefix_bytes: &[u8]| {
      let iter: Box<dyn SeekableRows + 'store_lt> =
        if cf_name_for_prefix == rocksdb::DEFAULT_COLUMN_FAMILY_NAME {
          Box::new(self.db.prefix_iterator(prefix_bytes))
        } else {
          let handle = self.get_cf_handle(&cf_name_for_prefix)?;
          Box::new(self.db.prefix_iterator_cf(&handle, prefix_bytes))
        };
      Ok(iter)
    });

    IterationHelper::new(config, general_iterator_factory, prefix_iterator_factory).execute()
  }

  fn find_by_prefix<Key, Val>(&self, cf_name: &str, prefix: &Key, direction: Direction) -> StoreResult<Vec<(Key, Val)>>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug + Clone,
    Val: DeserializeOwned + Debug,
  {
    let iter_config = IterConfig::new_deserializing(
      cf_name.to_string(),
      Some(prefix.clone()),                    // SerKey is Key (from prefix.clone())
      None,                                    // start
      matches!(direction, Direction::Reverse), // reverse
      None,                                    // control
      Box::new(|k_bytes, v_bytes| deserialize_kv(k_bytes, v_bytes)), // deserializer
    );

    match self.iterate::<Key, Key, Val>(iter_config)? {
      IterationResult::DeserializedItems(iter) => iter.collect(),
      _ => Err(StoreError::Other("find_by_prefix: Expected DeserializedItems".into())),
    }
  }

  fn find_from<Key, Val, F>(
    &self,
    cf_name: &str,
    start_key: Key,
    direction: Direction,
    control_fn: F,
  ) -> StoreResult<Vec<(Key, Val)>>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug,
    Val: DeserializeOwned + Debug,
    F: FnMut(&[u8], &[u8], usize) -> IterationControlDecision + 'static,
  {
    let iter_config = IterConfig::new_deserializing(
      cf_name.to_string(),
      None,                                                          // prefix
      Some(start_key),                                               // SerKey is Key (from start_key)
      matches!(direction, Direction::Reverse),                       // reverse
      Some(Box::new(control_fn)),                                    // control
      Box::new(|k_bytes, v_bytes| deserialize_kv(k_bytes, v_bytes)), // deserializer
    );

    match self.iterate::<Key, Key, Val>(iter_config)? {
      IterationResult::DeserializedItems(iter) => iter.collect(),
      _ => Err(StoreError::Other("find_from: Expected DeserializedItems".into())),
    }
  }

  fn find_from_with_expire_val<Key, Val, F>(
    &self,
    cf_name: &str,
    start: &Key,
    reverse: bool,
    control_fn: F,
  ) -> Result<Vec<(Key, ValueWithExpiry<Val>)>, String>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug + Clone,
    Val: DeserializeOwned + Debug,
    F: FnMut(&[u8], &[u8], usize) -> IterationControlDecision + 'static,
  {
    let iter_config = IterConfig::new_deserializing(
      cf_name.to_string(),
      None,                                                                 // prefix
      Some(start.clone()),                                                  // SerKey is Key (from start.clone())
      reverse,                                                              // reverse
      Some(Box::new(control_fn)),                                           // control
      Box::new(|k_bytes, v_bytes| deserialize_kv_expiry(k_bytes, v_bytes)), // deserializer
    );

    match self.iterate::<Key, Key, ValueWithExpiry<Val>>(iter_config) {
      Ok(IterationResult::DeserializedItems(iter)) => iter.collect::<Result<_, _>>().map_err(|e| e.to_string()),
      Ok(_) => Err("find_from_with_expire_val: Expected DeserializedItems from iteration".to_string()),
      Err(e) => Err(e.to_string()),
    }
  }

  fn find_by_prefix_with_expire_val<Key, Val, F>(
    &self,
    cf_name: &str,
    prefix_key: &Key,
    reverse: bool,
    control_fn: F,
  ) -> Result<Vec<(Key, ValueWithExpiry<Val>)>, String>
  where
    Key: ByteDecodable + AsBytes + DeserializeOwned + Hash + Eq + PartialEq + Debug + Clone,
    Val: DeserializeOwned + Debug,
    F: FnMut(&[u8], &[u8], usize) -> IterationControlDecision + 'static,
  {
    let iter_config = IterConfig::new_deserializing(
      cf_name.to_string(),
      Some(prefix_key.clone()),   // SerKey is Key (from prefix_key.clone())
      None,                       // start
      reverse,                    // reverse
      Some(Box::new(control_fn)), // control
      Box::new(|k_bytes, v_bytes| deserialize_kv_expiry(k_bytes, v_bytes)), // deserializer
    );

    match self.iterate::<Key, Key, ValueWithExpiry<Val>>(iter_config) {
      Ok(IterationResult::DeserializedItems(iter)) => iter.collect::<Result<_, _>>().map_err(|e| e.to_string()),
      Ok(_) => Err("find_by_prefix_with_expire_val: Expected DeserializedItems from iteration".to_string()),
      Err(e) => Err(e.to_string()),
    }
  }
}