use lmdb::{
Cursor, Database, DatabaseFlags, Environment, RoTransaction, RwTransaction, Transaction,
WriteFlags,
};
use lmdb_sys::{MDB_NEXT, MDB_NEXT_DUP, MDB_SET_RANGE};
use uuid::Uuid;
use wm_core::{CoreError, Galaxy, Result};
use crate::Memory;
pub const IDX_CONTENT_HASH: &str = "idx_content_hash";
pub const IDX_TAGS: &str = "idx_tags";
pub const IDX_IMPORTANCE: &str = "idx_importance";
pub const IDX_TEMPORAL: &str = "idx_temporal";
pub const INDEX_DBS: &[(&str, DatabaseFlags)] = &[
(IDX_CONTENT_HASH, DatabaseFlags::empty()),
(IDX_TAGS, DatabaseFlags::DUP_SORT),
(IDX_IMPORTANCE, DatabaseFlags::DUP_SORT),
(IDX_TEMPORAL, DatabaseFlags::DUP_SORT),
];
#[derive(Clone, Copy)]
pub struct IndexDbs {
content_hash: Database,
tags: Database,
importance: Database,
temporal: Database,
}
impl IndexDbs {
pub fn open(env: &Environment) -> Result<Self> {
Ok(Self {
content_hash: open_db(env, IDX_CONTENT_HASH)?,
tags: open_db(env, IDX_TAGS)?,
importance: open_db(env, IDX_IMPORTANCE)?,
temporal: open_db(env, IDX_TEMPORAL)?,
})
}
pub fn add(&self, tx: &mut RwTransaction, galaxy: Galaxy, memory: &Memory) -> Result<()> {
let id_bytes = memory.metadata.id.as_bytes();
let key = index_key(galaxy, memory.metadata.content_hash.as_bytes());
tx.put(self.content_hash, &key, id_bytes, WriteFlags::default())
.map_err(|e| CoreError::Memory(format!("idx_content_hash put: {e}")))?;
for tag in &memory.metadata.tags {
let key = index_key(galaxy, tag.as_bytes());
tx.put(self.tags, &key, id_bytes, WriteFlags::default())
.map_err(|e| CoreError::Memory(format!("idx_tags put: {e}")))?;
}
let imp_bytes = encode_f32(memory.metadata.importance);
let key = index_key(galaxy, &imp_bytes);
tx.put(self.importance, &key, id_bytes, WriteFlags::default())
.map_err(|e| CoreError::Memory(format!("idx_importance put: {e}")))?;
let ts_bytes = encode_timestamp(memory.metadata.created_at);
let key = index_key(galaxy, &ts_bytes);
tx.put(self.temporal, &key, id_bytes, WriteFlags::default())
.map_err(|e| CoreError::Memory(format!("idx_temporal put: {e}")))?;
Ok(())
}
pub fn remove(&self, tx: &mut RwTransaction, galaxy: Galaxy, memory: &Memory) -> Result<()> {
let key = index_key(galaxy, memory.metadata.content_hash.as_bytes());
let _ = tx.del(self.content_hash, &key, None);
for tag in &memory.metadata.tags {
let key = index_key(galaxy, tag.as_bytes());
let _ = tx.del(self.tags, &key, None);
}
let imp_bytes = encode_f32(memory.metadata.importance);
let key = index_key(galaxy, &imp_bytes);
let _ = tx.del(self.importance, &key, None);
let ts_bytes = encode_timestamp(memory.metadata.created_at);
let key = index_key(galaxy, &ts_bytes);
let _ = tx.del(self.temporal, &key, None);
Ok(())
}
pub fn find_by_content_hash(
&self,
tx: &RoTransaction,
galaxy: Galaxy,
hash: &str,
) -> Result<Option<Uuid>> {
let key = index_key(galaxy, hash.as_bytes());
match tx.get(self.content_hash, &key) {
Ok(bytes) => {
let id = decode_uuid(bytes)?;
Ok(Some(id))
}
Err(lmdb::Error::NotFound) => Ok(None),
Err(e) => Err(CoreError::Memory(format!("idx_content_hash get: {e}"))),
}
}
pub fn find_by_tag(&self, tx: &RoTransaction, galaxy: Galaxy, tag: &str) -> Result<Vec<Uuid>> {
let start_key = index_key(galaxy, tag.as_bytes());
let cursor = tx
.open_ro_cursor(self.tags)
.map_err(|e| CoreError::Memory(format!("idx_tags cursor: {e}")))?;
let mut ids = Vec::new();
match cursor.get(Some(&start_key), None, MDB_SET_RANGE) {
Ok((key_opt, val)) => {
let key_matches = key_opt.is_none_or(|k| k == start_key.as_slice());
if key_matches {
if let Ok(id) = decode_uuid(val) {
ids.push(id);
}
while let Ok((_, val)) = cursor.get(None, None, MDB_NEXT_DUP) {
if let Ok(id) = decode_uuid(val) {
ids.push(id);
}
}
}
}
Err(lmdb::Error::NotFound) => {}
Err(e) => return Err(CoreError::Memory(format!("idx_tags cursor get: {e}"))),
}
drop(cursor);
Ok(ids)
}
pub fn find_by_importance_range(
&self,
tx: &RoTransaction,
galaxy: Galaxy,
min: f32,
max: f32,
) -> Result<Vec<Uuid>> {
let prefix = galaxy_prefix(galaxy);
let start_key = index_key(galaxy, &encode_f32(min));
let max_bytes = encode_f32(max);
let cursor = tx
.open_ro_cursor(self.importance)
.map_err(|e| CoreError::Memory(format!("idx_importance cursor: {e}")))?;
let mut ids = Vec::new();
let mut current = cursor.get(Some(&start_key), None, MDB_SET_RANGE).ok();
while let Some((key_opt, val)) = current {
let key = key_opt.unwrap_or(&start_key);
if !key.starts_with(&prefix) {
break;
}
let value_bytes = &key[prefix.len()..];
if value_bytes > max_bytes.as_slice() {
break;
}
if let Ok(id) = decode_uuid(val) {
ids.push(id);
}
current = cursor.get(None, None, MDB_NEXT).ok();
}
drop(cursor);
Ok(ids)
}
pub fn find_by_time_range(
&self,
tx: &RoTransaction,
galaxy: Galaxy,
after: chrono::DateTime<chrono::Utc>,
before: chrono::DateTime<chrono::Utc>,
) -> Result<Vec<Uuid>> {
let prefix = galaxy_prefix(galaxy);
let start_key = index_key(galaxy, &encode_timestamp(after));
let max_bytes = encode_timestamp(before);
let cursor = tx
.open_ro_cursor(self.temporal)
.map_err(|e| CoreError::Memory(format!("idx_temporal cursor: {e}")))?;
let mut ids = Vec::new();
let mut current = cursor.get(Some(&start_key), None, MDB_SET_RANGE).ok();
while let Some((key_opt, val)) = current {
let key = key_opt.unwrap_or(&start_key);
if !key.starts_with(&prefix) {
break;
}
let value_bytes = &key[prefix.len()..];
if value_bytes > max_bytes.as_slice() {
break;
}
if let Ok(id) = decode_uuid(val) {
ids.push(id);
}
current = cursor.get(None, None, MDB_NEXT).ok();
}
drop(cursor);
Ok(ids)
}
}
fn open_db(env: &Environment, name: &str) -> Result<Database> {
env.open_db(Some(name))
.map_err(|e| CoreError::Memory(format!("LMDB open_db {name}: {e}")))
}
fn galaxy_prefix(galaxy: Galaxy) -> Vec<u8> {
let name = galaxy.db_name();
let mut key = Vec::with_capacity(name.len() + 1);
key.extend_from_slice(name.as_bytes());
key.push(0);
key
}
fn index_key(galaxy: Galaxy, value_bytes: &[u8]) -> Vec<u8> {
let mut key = galaxy_prefix(galaxy);
key.extend_from_slice(value_bytes);
key
}
const fn encode_f32(value: f32) -> [u8; 4] {
value.to_bits().to_be_bytes()
}
const fn encode_timestamp(ts: chrono::DateTime<chrono::Utc>) -> [u8; 8] {
ts.timestamp().to_be_bytes()
}
fn decode_uuid(bytes: &[u8]) -> Result<Uuid> {
Uuid::from_slice(bytes).map_err(|e| CoreError::Memory(format!("UUID decode: {e}")))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{Memory, MemoryStore};
use tempfile::tempdir;
use wm_core::Galaxy;
fn setup() -> (tempfile::TempDir, MemoryStore) {
let tmp = tempdir().unwrap();
let store = MemoryStore::open_default(tmp.path()).unwrap();
(tmp, store)
}
#[test]
fn content_hash_index_o1_lookup() {
let (_tmp, store) = setup();
let mem = Memory::new(Galaxy::Codex, "hello world".into());
let id = mem.metadata.id;
let hash = mem.metadata.content_hash.clone();
store.put(Galaxy::Codex, &mem).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
let found = store
.index_dbs()
.find_by_content_hash(&tx, Galaxy::Codex, &hash)
.unwrap();
tx.commit().unwrap();
assert_eq!(found, Some(id));
}
#[test]
fn content_hash_index_miss() {
let (_tmp, store) = setup();
let tx = store.env().begin_ro_txn().unwrap();
let found = store
.index_dbs()
.find_by_content_hash(&tx, Galaxy::Codex, "nonexistent")
.unwrap();
tx.commit().unwrap();
assert!(found.is_none());
}
#[test]
fn tag_index_returns_all_tagged() {
let (_tmp, store) = setup();
let mem1 = Memory::new(Galaxy::Codex, "a".into()).with_tags(vec!["rust".into()]);
let mem2 = Memory::new(Galaxy::Codex, "b".into()).with_tags(vec!["rust".into()]);
let mem3 = Memory::new(Galaxy::Codex, "c".into()).with_tags(vec!["python".into()]);
let id1 = mem1.metadata.id;
let id2 = mem2.metadata.id;
store.put(Galaxy::Codex, &mem1).unwrap();
store.put(Galaxy::Codex, &mem2).unwrap();
store.put(Galaxy::Codex, &mem3).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
let rust_ids = store
.index_dbs()
.find_by_tag(&tx, Galaxy::Codex, "rust")
.unwrap();
tx.commit().unwrap();
assert_eq!(rust_ids.len(), 2);
assert!(rust_ids.contains(&id1));
assert!(rust_ids.contains(&id2));
}
#[test]
fn tag_index_galaxy_scoped() {
let (_tmp, store) = setup();
let mem1 = Memory::new(Galaxy::Codex, "a".into()).with_tags(vec!["shared".into()]);
let mem2 = Memory::new(Galaxy::Research, "b".into()).with_tags(vec!["shared".into()]);
store.put(Galaxy::Codex, &mem1).unwrap();
store.put(Galaxy::Research, &mem2).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
let codex_ids = store
.index_dbs()
.find_by_tag(&tx, Galaxy::Codex, "shared")
.unwrap();
let research_ids = store
.index_dbs()
.find_by_tag(&tx, Galaxy::Research, "shared")
.unwrap();
tx.commit().unwrap();
assert_eq!(codex_ids.len(), 1);
assert_eq!(research_ids.len(), 1);
}
#[test]
fn importance_range_query() {
let (_tmp, store) = setup();
store
.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, "low".into()).with_importance(0.1),
)
.unwrap();
let mid = Memory::new(Galaxy::Codex, "mid".into()).with_importance(0.5);
let mid_id = mid.metadata.id;
store.put(Galaxy::Codex, &mid).unwrap();
store
.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, "high".into()).with_importance(0.9),
)
.unwrap();
let tx = store.env().begin_ro_txn().unwrap();
let ids = store
.index_dbs()
.find_by_importance_range(&tx, Galaxy::Codex, 0.4, 0.6)
.unwrap();
tx.commit().unwrap();
assert_eq!(ids.len(), 1);
assert_eq!(ids[0], mid_id);
}
#[test]
fn importance_range_query_full_range() {
let (_tmp, store) = setup();
for i in 0..10 {
let imp = i as f32 * 0.1;
store
.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, format!("m{i}")).with_importance(imp),
)
.unwrap();
}
let tx = store.env().begin_ro_txn().unwrap();
let ids = store
.index_dbs()
.find_by_importance_range(&tx, Galaxy::Codex, 0.0, 1.0)
.unwrap();
tx.commit().unwrap();
assert_eq!(ids.len(), 10);
}
#[test]
fn temporal_range_query() {
let (_tmp, store) = setup();
let t0 = chrono::Utc::now();
std::thread::sleep(std::time::Duration::from_millis(10));
let mid = Memory::new(Galaxy::Codex, "mid".into());
let mid_id = mid.metadata.id;
store.put(Galaxy::Codex, &mid).unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
let t2 = chrono::Utc::now();
let tx = store.env().begin_ro_txn().unwrap();
let ids = store
.index_dbs()
.find_by_time_range(&tx, Galaxy::Codex, t0, t2)
.unwrap();
tx.commit().unwrap();
assert_eq!(ids.len(), 1);
assert_eq!(ids[0], mid_id);
}
#[test]
fn delete_removes_index_entries() {
let (_tmp, store) = setup();
let mem = Memory::new(Galaxy::Codex, "test".into())
.with_tags(vec!["tag1".into()])
.with_importance(0.7);
let id = mem.metadata.id;
let hash = mem.metadata.content_hash.clone();
store.put(Galaxy::Codex, &mem).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
assert!(
store
.index_dbs()
.find_by_content_hash(&tx, Galaxy::Codex, &hash)
.unwrap()
.is_some()
);
assert_eq!(
store
.index_dbs()
.find_by_tag(&tx, Galaxy::Codex, "tag1")
.unwrap()
.len(),
1
);
tx.commit().unwrap();
store.delete(Galaxy::Codex, id).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
assert!(
store
.index_dbs()
.find_by_content_hash(&tx, Galaxy::Codex, &hash)
.unwrap()
.is_none()
);
assert_eq!(
store
.index_dbs()
.find_by_tag(&tx, Galaxy::Codex, "tag1")
.unwrap()
.len(),
0
);
tx.commit().unwrap();
}
#[test]
fn put_batch_updates_indexes() {
let (_tmp, store) = setup();
let memories: Vec<Memory> = (0..5)
.map(|i| {
Memory::new(Galaxy::Codex, format!("batch-{i}"))
.with_tags(vec![format!("tag{i}")])
.with_importance(i as f32 * 0.2)
})
.collect();
store.put_batch(Galaxy::Codex, &memories).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
for i in 0..5 {
let ids = store
.index_dbs()
.find_by_tag(&tx, Galaxy::Codex, &format!("tag{i}"))
.unwrap();
assert_eq!(ids.len(), 1, "tag{i} should have 1 entry");
}
tx.commit().unwrap();
}
#[test]
fn find_by_content_hash_indexed_matches_scan() {
let (_tmp, store) = setup();
let mem = Memory::new(Galaxy::Codex, "dedup test".into());
let id = mem.metadata.id;
let hash = mem.metadata.content_hash.clone();
store.put(Galaxy::Codex, &mem).unwrap();
let tx = store.env().begin_ro_txn().unwrap();
let indexed = store
.index_dbs()
.find_by_content_hash(&tx, Galaxy::Codex, &hash)
.unwrap();
tx.commit().unwrap();
let scanned = store
.find_by_content_hash_scan(Galaxy::Codex, &hash)
.unwrap();
assert_eq!(indexed, scanned);
assert_eq!(indexed, Some(id));
}
#[test]
fn key_encoding_sorts_correctly() {
let values = [0.0_f32, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0];
let encoded: Vec<[u8; 4]> = values.map(encode_f32).to_vec();
for i in 0..encoded.len() - 1 {
assert!(
encoded[i] < encoded[i + 1],
"f32 sort order broken: {:?} >= {:?}",
values[i],
values[i + 1]
);
}
}
}