rocksolid 3.0.0

mod common;

use common::setup_logging;
use rocksolid::cf_store::{CFOperations, RocksDbCFStore};
use rocksolid::config::{BaseCfConfig, RockSolidComparatorOpt, RocksDbCFStoreConfig};
use rocksolid::error::StoreResult;
use rocksolid::iter::{IterConfig, IterationResult};
use std::collections::HashMap;
use tempfile::TempDir;

const DEFAULT_CF_NAME: &str = rocksdb::DEFAULT_COLUMN_FAMILY_NAME;
const SORTED_CF: &str = "sorted_cf";

// Bytewise ordering puts "item10" before "item2"; both natural comparators must not.
const UNSORTED_KEYS: [&str; 5] = ["item10", "item2", "item1", "item20", "item3"];
const NATURAL_ORDER: [&str; 5] = ["item1", "item2", "item3", "item10", "item20"];

fn config_with_comparator(path: String, comparator: RockSolidComparatorOpt) -> RocksDbCFStoreConfig {
  let mut cf_configs = HashMap::new();
  cf_configs.insert(DEFAULT_CF_NAME.to_string(), BaseCfConfig::default());
  cf_configs.insert(
    SORTED_CF.to_string(),
    BaseCfConfig {
      comparator: Some(comparator),
      ..Default::default()
    },
  );

  RocksDbCFStoreConfig {
    path,
    create_if_missing: true,
    column_families_to_open: vec![DEFAULT_CF_NAME.to_string(), SORTED_CF.to_string()],
    column_family_configs: cf_configs,
    ..Default::default()
  }
}

fn keys_in_order(store: &RocksDbCFStore) -> StoreResult<Vec<String>> {
  let config: IterConfig<'_, String, Vec<u8>, Vec<u8>> =
    IterConfig::new_raw(SORTED_CF.to_string(), None, None, false, None);
  match store.iterate(config)? {
    IterationResult::RawItems(iter) => iter
      .map(|res| res.map(|(k, _)| String::from_utf8_lossy(&k).into_owned()))
      .collect(),
    _ => unreachable!("new_raw yields RawItems"),
  }
}

/// RocksDB records the comparator name in the CF's OPTIONS file and refuses to
/// reopen a CF whose recorded name does not match the one supplied at open time,
/// so ordering is asserted both on the original handle and after a reopen.
fn assert_natural_ordering_survives_reopen(test_name: &str, comparator: RockSolidComparatorOpt) {
  setup_logging();
  let temp_dir = TempDir::new().unwrap();
  let path = temp_dir.path().join(test_name).to_str().unwrap().to_string();

  {
    let store = RocksDbCFStore::open(config_with_comparator(path.clone(), comparator.clone())).unwrap();
    for key in UNSORTED_KEYS {
      store.put(SORTED_CF, key, &format!("value_for_{key}")).unwrap();
    }
    assert_eq!(keys_in_order(&store).unwrap(), NATURAL_ORDER);
  }

  let reopened = RocksDbCFStore::open(config_with_comparator(path, comparator))
    .expect("reopening with the same comparator must succeed");
  assert_eq!(keys_in_order(&reopened).unwrap(), NATURAL_ORDER);

  let value: Option<String> = reopened.get(SORTED_CF, "item10").unwrap();
  assert_eq!(value.as_deref(), Some("value_for_item10"));
}

#[test]
#[cfg(feature = "natlex_sort")]
fn natural_lexicographical_ordering_survives_reopen() {
  assert_natural_ordering_survives_reopen(
    "natlex_reopen_db",
    RockSolidComparatorOpt::NaturalLexicographical { ignore_case: false },
  );
}

#[test]
#[cfg(feature = "nat_sort")]
fn natural_ordering_survives_reopen() {
  assert_natural_ordering_survives_reopen("natsort_reopen_db", RockSolidComparatorOpt::Natural { ignore_case: false });
}

#[test]
fn bytewise_default_orders_lexicographically() {
  setup_logging();
  let temp_dir = TempDir::new().unwrap();
  let path = temp_dir.path().join("bytewise_db").to_str().unwrap().to_string();

  let store = RocksDbCFStore::open(config_with_comparator(path, RockSolidComparatorOpt::None)).unwrap();
  for key in UNSORTED_KEYS {
    store.put(SORTED_CF, key, &format!("value_for_{key}")).unwrap();
  }

  assert_eq!(
    keys_in_order(&store).unwrap(),
    vec!["item1", "item10", "item2", "item20", "item3"]
  );
}