use std::{
collections::BTreeMap,
ops::ControlFlow,
sync::{
Arc,
atomic::{AtomicI64, Ordering},
},
time::Duration,
};
use hidane_proto::google::firestore::v1::{Value, value::ValueType};
use super::MemoryStore;
use crate::{
path::ResourcePath,
store::{ListItem, ListOptions, ReadTime, Store, StoreError, StoredDocument},
};
const DB: &str = "projects/demo/databases/(default)";
fn path(p: &str) -> ResourcePath {
ResourcePath::parse(p).unwrap()
}
fn int(i: i64) -> Value {
Value {
value_type: Some(ValueType::IntegerValue(i)),
}
}
fn fields(n: i64) -> BTreeMap<String, Value> {
[("n".to_owned(), int(n))].into()
}
fn n_of(doc: &StoredDocument) -> i64 {
match doc.fields()["n"].value_type {
Some(ValueType::IntegerValue(n)) => n,
_ => panic!("no n"),
}
}
fn store_with_clock(start: i64) -> (MemoryStore, Arc<AtomicI64>) {
let now = Arc::new(AtomicI64::new(start));
let clock = now.clone();
(
MemoryStore::with_clock(move || clock.load(Ordering::SeqCst)),
now,
)
}
fn set(store: &MemoryStore, p: &str, n: i64) -> ReadTime {
store
.commit(DB, &mut |batch| {
batch.set(&path(p), fields(n));
Ok(())
})
.unwrap()
.commit_time
}
fn delete(store: &MemoryStore, p: &str) -> ReadTime {
store
.commit(DB, &mut |batch| {
batch.delete(&path(p));
Ok(())
})
.unwrap()
.commit_time
}
fn collect(
scan: impl FnOnce(&mut dyn FnMut(&Arc<StoredDocument>) -> ControlFlow<()>),
) -> Vec<String> {
let mut names = Vec::new();
scan(&mut |doc| {
names.push(doc.path.to_string());
ControlFlow::Continue(())
});
names
}
#[test]
fn reads_see_the_version_at_their_read_time() {
let (store, now) = store_with_clock(1_000);
let t1 = set(&store, "users/alice", 1);
now.store(2_000, Ordering::SeqCst);
let t2 = set(&store, "users/alice", 2);
now.store(3_000, Ordering::SeqCst);
let t3 = delete(&store, "users/alice");
let alice = path("users/alice");
assert_eq!(store.get(DB, &alice, ReadTime(t1.0 - 1)), None);
let v1 = store.get(DB, &alice, t1).unwrap();
let v2 = store.get(DB, &alice, ReadTime(t3.0 - 1)).unwrap();
assert_eq!((n_of(&v1), n_of(&v2)), (1, 2));
assert_eq!(store.get(DB, &alice, t3), None);
assert_eq!(store.get(DB, &alice, ReadTime::MAX), None);
assert_eq!(v1.create_time, t1.to_timestamp());
assert_eq!(v2.create_time, t1.to_timestamp());
assert_eq!(v2.update_time, t2.to_timestamp());
}
#[test]
fn commits_report_what_changed() {
let store = MemoryStore::new();
set(&store, "users/alice", 1);
let commit = store
.commit(DB, &mut |batch| {
batch.set(&path("users/alice"), fields(2));
batch.set(&path("users/bob"), fields(1));
batch.delete(&path("users/carol")); Ok(())
})
.unwrap();
let summary: Vec<_> = commit
.changes
.iter()
.map(|c| {
(
c.path.to_string(),
c.before.as_deref().map(n_of),
c.after.as_deref().map(n_of),
)
})
.collect();
assert_eq!(
summary,
[
("users/alice".to_owned(), Some(1), Some(2)),
("users/bob".to_owned(), None, Some(1)),
]
);
}
#[test]
fn batch_reads_its_own_writes() {
let store = MemoryStore::new();
set(&store, "users/alice", 1);
store
.commit(DB, &mut |batch| {
assert_eq!(n_of(&batch.get(&path("users/alice")).unwrap()), 1);
batch.set(&path("users/alice"), fields(2));
assert_eq!(n_of(&batch.get(&path("users/alice")).unwrap()), 2);
batch.delete(&path("users/alice"));
assert!(batch.get(&path("users/alice")).is_none());
batch.set(&path("users/alice"), fields(3));
Ok(())
})
.unwrap();
let alice = store.get(DB, &path("users/alice"), ReadTime::MAX).unwrap();
assert_eq!(n_of(&alice), 3);
}
#[test]
fn a_failed_commit_applies_nothing() {
let store = MemoryStore::new();
set(&store, "users/alice", 1);
let err = store
.commit(DB, &mut |batch| {
batch.set(&path("users/alice"), fields(99));
batch.set(&path("users/bob"), fields(99));
Err(StoreError::FailedPrecondition("no".into()))
})
.unwrap_err();
assert_eq!(err, StoreError::FailedPrecondition("no".into()));
let alice = store.get(DB, &path("users/alice"), ReadTime::MAX).unwrap();
assert_eq!(n_of(&alice), 1);
assert!(store.get(DB, &path("users/bob"), ReadTime::MAX).is_none());
}
#[test]
fn commit_times_increase_even_with_a_frozen_clock() {
let (store, _) = store_with_clock(5_000);
let times: Vec<i64> = (0..5).map(|i| set(&store, "c/d", i).0).collect();
assert_eq!(times, [5_000, 5_001, 5_002, 5_003, 5_004]);
assert_eq!(store.latest_read_time(DB), ReadTime(5_004));
let latest = store
.get(DB, &path("c/d"), store.latest_read_time(DB))
.unwrap();
assert_eq!(n_of(&latest), 4);
}
#[test]
fn collection_scans_skip_subcollections_and_follow_name_order() {
let store = MemoryStore::new();
for p in [
"c/b",
"c/a/sub/x",
"c/a",
"c/__id10__",
"c/__id5__",
"c/\u{1f600}",
"c/\u{ff5e}",
"c/aa/deeper/y",
"c/aa",
"c-b/x",
"other/z",
] {
set(&store, p, 0);
}
let names = collect(|visit| store.scan_collection(DB, &path("c"), ReadTime::MAX, visit));
assert_eq!(
names,
[
"c/__id5__",
"c/__id10__",
"c/a",
"c/aa",
"c/b",
"c/\u{ff5e}",
"c/\u{1f600}"
]
);
let sub = collect(|visit| store.scan_collection(DB, &path("c/a/sub"), ReadTime::MAX, visit));
assert_eq!(sub, ["c/a/sub/x"]);
}
#[test]
fn scans_are_snapshots_and_can_stop_early() {
let (store, now) = store_with_clock(1_000);
let t1 = set(&store, "c/a", 1);
now.store(2_000, Ordering::SeqCst);
set(&store, "c/b", 1);
delete(&store, "c/a");
assert_eq!(
collect(|v| store.scan_collection(DB, &path("c"), t1, v)),
["c/a"]
);
assert_eq!(
collect(|v| store.scan_collection(DB, &path("c"), ReadTime::MAX, v)),
["c/b"]
);
set(&store, "c/c", 1);
let mut seen = 0;
store.scan_collection(DB, &path("c"), ReadTime::MAX, &mut |_| {
seen += 1;
ControlFlow::Break(())
});
assert_eq!(seen, 1);
}
#[test]
fn collection_groups_cover_every_depth_in_full_path_order() {
let store = MemoryStore::new();
for p in [
"items/top",
"c/b/items/x",
"c/a/items/y",
"c/a/items/x",
"c/a/sub/b/items/x",
"c-b/x/items/x",
"c/a/other/x",
] {
set(&store, p, 0);
}
let all = collect(|v| {
store.scan_collection_group(DB, &ResourcePath::root(), "items", ReadTime::MAX, v);
});
assert_eq!(
all,
[
"c/a/items/x",
"c/a/items/y",
"c/a/sub/b/items/x",
"c/b/items/x",
"c-b/x/items/x",
"items/top",
]
);
let under_c_a = collect(|v| {
store.scan_collection_group(DB, &path("c/a"), "items", ReadTime::MAX, v);
});
assert_eq!(
under_c_a,
["c/a/items/x", "c/a/items/y", "c/a/sub/b/items/x"]
);
set(&store, "items/top/items/child", 0);
let under_items_top = collect(|v| {
store.scan_collection_group(DB, &path("items/top"), "items", ReadTime::MAX, v);
});
assert_eq!(under_items_top, ["items/top/items/child"]);
}
#[test]
fn descendants_cover_every_depth_but_not_the_parent() {
let store = MemoryStore::new();
for p in ["c/a", "c/a/s/x", "c/a/s/x/t/y", "c/ab/s/z", "c/b", "d/a"] {
set(&store, p, 0);
}
let under_c_a = collect(|v| {
store.scan_descendants(DB, &path("c/a"), ReadTime::MAX, v);
});
assert_eq!(under_c_a, ["c/a/s/x", "c/a/s/x/t/y"]);
let all = collect(|v| {
store.scan_descendants(DB, &ResourcePath::root(), ReadTime::MAX, v);
});
assert_eq!(
all,
["c/a", "c/a/s/x", "c/a/s/x/t/y", "c/ab/s/z", "c/b", "d/a"]
);
}
#[test]
fn collection_ids_include_collections_with_only_nested_documents() {
let (store, now) = store_with_clock(1_000);
set(&store, "users/alice", 1);
set(&store, "users/alice/posts/p1", 1);
set(&store, "users/alice/drafts/d1", 1);
set(&store, "ghosts/g/x/y", 1);
let t = set(&store, "tmp/t", 1);
now.store(2_000, Ordering::SeqCst);
delete(&store, "tmp/t");
assert_eq!(
store.list_collection_ids(DB, &ResourcePath::root(), ReadTime::MAX),
["ghosts", "users"]
);
assert_eq!(
store.list_collection_ids(DB, &ResourcePath::root(), t),
["ghosts", "tmp", "users"]
);
assert_eq!(
store.list_collection_ids(DB, &path("users/alice"), ReadTime::MAX),
["drafts", "posts"]
);
assert!(
store
.list_collection_ids(DB, &path("users/bob"), ReadTime::MAX)
.is_empty()
);
}
#[test]
fn versions_older_than_the_retention_window_are_dropped() {
let (store, now) = store_with_clock(100);
let store = store.with_retention(Duration::from_micros(10));
set(&store, "c/d", 1); now.store(200, Ordering::SeqCst);
set(&store, "c/d", 2); assert_eq!(
n_of(&store.get(DB, &path("c/d"), ReadTime(150)).unwrap()),
1
);
now.store(300, Ordering::SeqCst);
set(&store, "c/d", 3); assert_eq!(store.get(DB, &path("c/d"), ReadTime(150)), None);
assert_eq!(
n_of(&store.get(DB, &path("c/d"), ReadTime(250)).unwrap()),
2
);
assert_eq!(
n_of(&store.get(DB, &path("c/d"), ReadTime(300)).unwrap()),
3
);
}
#[test]
fn databases_are_isolated_and_clear_drops_everything() {
let store = MemoryStore::new();
set(&store, "c/d", 1);
let other = "projects/demo/databases/second";
assert!(store.get(other, &path("c/d"), ReadTime::MAX).is_none());
store.clear();
assert!(store.get(DB, &path("c/d"), ReadTime::MAX).is_none());
}
#[test]
fn read_time_round_trips_through_timestamps() {
for micros in [
0,
1,
999_999,
1_000_000,
-1,
-1_000_001,
1_760_000_000_123_456,
] {
assert_eq!(
ReadTime::from_timestamp(&ReadTime(micros).to_timestamp()),
ReadTime(micros)
);
}
}
#[test]
fn listings_page_and_report_missing_documents() {
let (store, now) = store_with_clock(1_000);
for p in ["c/a", "c/b", "c/ghost/sub/x", "c/gone", "c/z/sub/y"] {
set(&store, p, 1);
}
set(&store, "c/z", 1);
now.store(2_000, Ordering::SeqCst);
delete(&store, "c/gone");
delete(&store, "c/z");
let list = |options: ListOptions<'_>| {
let mut out = Vec::new();
store.list_collection(DB, &path("c"), ReadTime::MAX, options, &mut |item| {
out.push(match item {
ListItem::Document(doc) => doc.path.to_string(),
ListItem::Missing(p) => format!("missing:{p}"),
});
ControlFlow::Continue(())
});
out
};
assert_eq!(list(ListOptions::default()), ["c/a", "c/b"]);
assert_eq!(
list(ListOptions {
include_missing: true,
..ListOptions::default()
}),
["c/a", "c/b", "missing:c/ghost", "missing:c/z"]
);
let b = path("c/b");
assert_eq!(
list(ListOptions {
after: Some(&b),
include_missing: true,
}),
["missing:c/ghost", "missing:c/z"]
);
}
#[test]
fn clearing_a_database_keeps_time_moving_and_other_databases() {
let (store, clock) = store_with_clock(1_000);
set(&store, "c/a", 1);
store
.commit("projects/p/databases/other", &mut |batch| {
batch.set(&path("c/b"), fields(2));
Ok(())
})
.unwrap();
let before = store.latest_read_time(DB);
clock.store(0, std::sync::atomic::Ordering::SeqCst); let cleared = store.clear_database(DB);
assert!(cleared > before);
assert!(store.get(DB, &path("c/a"), ReadTime::MAX).is_none());
assert!(
store.get(DB, &path("c/a"), before).is_none(),
"old versions are gone too"
);
assert!(
store
.get("projects/p/databases/other", &path("c/b"), ReadTime::MAX)
.is_some()
);
assert!(set(&store, "c/a", 3) > cleared);
let mut names = store.database_names();
names.sort();
assert_eq!(
names,
[DB.to_owned(), "projects/p/databases/other".to_owned()]
);
}