use nodedb_types::Namespace;
use nodedb_types::result::QueryResult;
use nodedb_types::value::Value;
use crate::error::LiteError;
use crate::storage::engine::{KvPair, StorageEngine, WriteOp};
type SnapshotEntry = (u8, Vec<u8>, Vec<u8>);
type TenantKeyTuple = (Vec<u8>, Vec<u8>, Namespace, Vec<u8>);
fn tenant_prefix(tenant_id: u64) -> Vec<u8> {
format!("t/{tenant_id}/").into_bytes()
}
const ALL_NAMESPACES: &[Namespace] = &[
Namespace::Meta,
Namespace::Vector,
Namespace::Graph,
Namespace::Crdt,
Namespace::LoroState,
Namespace::Spatial,
Namespace::Strict,
Namespace::Columnar,
Namespace::Kv,
Namespace::Array,
Namespace::ArrayOpLog,
Namespace::ArrayDelta,
Namespace::Fts,
];
async fn collect_tenant_keys<S: StorageEngine>(
storage: &S,
tenant_id: u64,
) -> Result<Vec<TenantKeyTuple>, LiteError> {
let explicit_prefix = tenant_prefix(tenant_id);
let mut result: Vec<TenantKeyTuple> = Vec::new();
for &ns in ALL_NAMESPACES {
let pairs: Vec<KvPair> = storage.scan_prefix(ns, b"").await?;
for (storage_key, value) in pairs {
if storage_key.starts_with(&explicit_prefix) {
let user_key = storage_key[explicit_prefix.len()..].to_vec();
result.push((storage_key, user_key, ns, value));
} else if tenant_id == 0 && !storage_key.starts_with(b"t/") {
let user_key = storage_key.clone();
result.push((storage_key, user_key, ns, value));
}
}
}
Ok(result)
}
pub async fn handle_create_tenant_snapshot<S: StorageEngine>(
storage: &S,
tenant_id: u64,
) -> Result<QueryResult, LiteError> {
let raw = collect_tenant_keys(storage, tenant_id).await?;
let entries: Vec<SnapshotEntry> = raw
.into_iter()
.map(|(_full_key, user_key, ns, value)| (ns as u8, user_key, value))
.collect();
let blob = zerompk::to_msgpack_vec(&entries).map_err(|e| LiteError::Storage {
detail: format!("tenant snapshot serialise failed: {e}"),
})?;
Ok(QueryResult {
columns: vec!["snapshot".into()],
rows: vec![vec![Value::Bytes(blob)]],
rows_affected: entries.len() as u64,
command: None,
})
}
pub async fn handle_restore_tenant_snapshot<S: StorageEngine>(
storage: &S,
tenant_id: u64,
snapshot: &[u8],
) -> Result<QueryResult, LiteError> {
let entries: Vec<SnapshotEntry> =
zerompk::from_msgpack(snapshot).map_err(|e| LiteError::Storage {
detail: format!("tenant snapshot parse failed: {e}"),
})?;
let prefix = tenant_prefix(tenant_id);
let mut ops: Vec<WriteOp> = Vec::with_capacity(entries.len());
for (ns_byte, user_key, value) in &entries {
let ns = Namespace::from_u8(*ns_byte).ok_or_else(|| LiteError::Storage {
detail: format!("snapshot contains unknown namespace byte {ns_byte}"),
})?;
let mut full_key = prefix.clone();
full_key.extend_from_slice(user_key);
ops.push(WriteOp::Put {
ns,
key: full_key,
value: value.clone(),
});
}
let written = ops.len();
storage.batch_write(&ops).await?;
Ok(QueryResult {
columns: Vec::new(),
rows: Vec::new(),
rows_affected: written as u64,
command: None,
})
}
pub async fn handle_purge_tenant<S: StorageEngine>(
storage: &S,
tenant_id: u64,
) -> Result<QueryResult, LiteError> {
let raw = collect_tenant_keys(storage, tenant_id).await?;
let ops: Vec<WriteOp> = raw
.into_iter()
.map(|(full_key, _user_key, ns, _value)| WriteOp::Delete { ns, key: full_key })
.collect();
let deleted = ops.len() as u64;
storage.batch_write(&ops).await?;
Ok(QueryResult {
columns: Vec::new(),
rows: Vec::new(),
rows_affected: deleted,
command: None,
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::pagedb_storage::PagedbStorageMem;
async fn make_storage() -> PagedbStorageMem {
PagedbStorageMem::open_in_memory().await.unwrap()
}
#[tokio::test]
async fn snapshot_roundtrip_tenant_zero_legacy_keys() {
let s = make_storage().await;
s.put(Namespace::Kv, b"doc:1", b"value1").await.unwrap();
s.put(Namespace::Kv, b"doc:2", b"value2").await.unwrap();
s.put(Namespace::Meta, b"schema:v1", b"meta").await.unwrap();
let result = handle_create_tenant_snapshot(&s, 0).await.unwrap();
assert_eq!(result.columns, vec!["snapshot"]);
assert_eq!(result.rows.len(), 1);
assert_eq!(result.rows_affected, 3);
let blob = match &result.rows[0][0] {
Value::Bytes(b) => b.clone(),
other => panic!("expected Bytes, got {other:?}"),
};
s.delete(Namespace::Kv, b"doc:1").await.unwrap();
s.delete(Namespace::Kv, b"doc:2").await.unwrap();
s.delete(Namespace::Meta, b"schema:v1").await.unwrap();
let restore = handle_restore_tenant_snapshot(&s, 0, &blob).await.unwrap();
assert_eq!(restore.rows_affected, 3);
let val = s.get(Namespace::Kv, b"t/0/doc:1").await.unwrap();
assert_eq!(val.as_deref(), Some(b"value1".as_slice()));
let val2 = s.get(Namespace::Kv, b"t/0/doc:2").await.unwrap();
assert_eq!(val2.as_deref(), Some(b"value2".as_slice()));
}
#[tokio::test]
async fn snapshot_roundtrip_non_zero_tenant() {
let s = make_storage().await;
s.put(Namespace::Strict, b"t/42/row:1", b"strict_data")
.await
.unwrap();
let result = handle_create_tenant_snapshot(&s, 42).await.unwrap();
assert_eq!(result.rows_affected, 1);
let blob = match &result.rows[0][0] {
Value::Bytes(b) => b.clone(),
other => panic!("expected Bytes, got {other:?}"),
};
let restore = handle_restore_tenant_snapshot(&s, 99, &blob).await.unwrap();
assert_eq!(restore.rows_affected, 1);
let val = s.get(Namespace::Strict, b"t/99/row:1").await.unwrap();
assert_eq!(val.as_deref(), Some(b"strict_data".as_slice()));
}
#[tokio::test]
async fn purge_tenant_removes_only_target_tenant() {
let s = make_storage().await;
s.put(Namespace::Kv, b"doc:1", b"v1").await.unwrap();
s.put(Namespace::Kv, b"doc:2", b"v2").await.unwrap();
s.put(Namespace::Kv, b"t/1/doc:3", b"v3").await.unwrap();
let result = handle_purge_tenant(&s, 0).await.unwrap();
assert_eq!(result.rows_affected, 2);
assert!(s.get(Namespace::Kv, b"doc:1").await.unwrap().is_none());
assert!(s.get(Namespace::Kv, b"doc:2").await.unwrap().is_none());
assert!(s.get(Namespace::Kv, b"t/1/doc:3").await.unwrap().is_some());
}
#[tokio::test]
async fn purge_tenant_is_idempotent() {
let s = make_storage().await;
s.put(Namespace::Kv, b"x", b"y").await.unwrap();
handle_purge_tenant(&s, 0).await.unwrap();
let second = handle_purge_tenant(&s, 0).await.unwrap();
assert_eq!(second.rows_affected, 0);
}
#[tokio::test]
async fn empty_tenant_snapshot_roundtrip() {
let s = make_storage().await;
let result = handle_create_tenant_snapshot(&s, 7).await.unwrap();
assert_eq!(result.rows_affected, 0);
assert_eq!(result.rows.len(), 1);
let blob = match &result.rows[0][0] {
Value::Bytes(b) => b.clone(),
other => panic!("expected Bytes, got {other:?}"),
};
let entries: Vec<SnapshotEntry> = zerompk::from_msgpack(&blob).unwrap();
assert!(entries.is_empty());
let restore = handle_restore_tenant_snapshot(&s, 7, &blob).await.unwrap();
assert_eq!(restore.rows_affected, 0);
}
#[tokio::test]
async fn tenant_zero_does_not_capture_other_tenants() {
let s = make_storage().await;
s.put(Namespace::Kv, b"my_key", b"val0").await.unwrap();
s.put(Namespace::Kv, b"t/5/key", b"val5").await.unwrap();
let result = handle_create_tenant_snapshot(&s, 0).await.unwrap();
assert_eq!(result.rows_affected, 1);
}
}