use serde_json::json;
use std::collections::BTreeMap;
use std::path::PathBuf;
use knot_types::RepoDid;
use serde_json::Value;
use super::*;
use crate::atproto::ServedRecord;
const COLLECTION: &str = "org.tangled.track.ticket";
fn repo(records: usize, versions: usize) -> ChainState {
let did = RepoDid::new("did:plc:abcdefghijklmnopqrstuvwx").unwrap();
let cid = knot_types::dag_cbor_cid(b"commit");
let version = |n: usize| -> Version {
serde_json::from_value(json!({
"cid": format!("bafyrei{n:052}"),
"changeId": format!("{n:040x}"),
"editor": "did:plc:alice",
"createdAt": "2024-05-01T00:00:00.000Z",
}))
.unwrap()
};
ChainState {
commit: cid,
rev: "3lubrptx57d22".into(),
collections: vec![COLLECTION.into()],
records: (0..records)
.map(|n| ServedRecord {
collection: COLLECTION.into(),
rkey: format!("r{n:03}"),
answer: json!({
"uri": format!("at://{did}/{COLLECTION}/r{n:03}"),
"cid": format!("bafyrei{n:052}"),
"value": { "text": n },
}),
})
.collect(),
versions: BTreeMap::from([(
format!("{COLLECTION}/r000"),
(0..versions).map(version).collect(),
)]),
car: Vec::new(),
did,
}
}
struct Emitted {
_dir: tempfile::TempDir,
rows: Rows,
root: PathBuf,
name: String,
}
fn emitted(repo: &ChainState) -> Emitted {
let dir = tempfile::tempdir().unwrap();
write_empty(dir.path()).unwrap();
let mut rows = Rows::paging();
let name = repo.dir_name();
let pooled = dir.path().join(&name);
let mut pool = Pool {
dir: &pooled,
verify: false,
files: BTreeMap::new(),
};
rows.add(&mut pool, repo, None).unwrap();
Emitted {
root: dir.path().to_path_buf(),
rows,
name,
_dir: dir,
}
}
impl Emitted {
fn find(rows: &[(String, String)], key: &str) -> Option<String> {
rows.iter()
.find(|(row, _)| row == key)
.map(|(_, file)| file.clone())
}
fn page(&self, rows: &[(String, String)], key: &str) -> Value {
let file = Self::find(rows, key).unwrap_or_else(|| panic!("no row for {key}"));
serde_json::from_slice(&std::fs::read(self.root.join(file)).unwrap()).unwrap()
}
fn list_records(
&self,
limit: &str,
reverse: &str,
collection: &str,
cursor: &str,
) -> Option<Value> {
let lrpage = Self::find(&self.rows.record_page, &format!("{limit}|{reverse}"))?;
let lrdir = Self::find(&self.rows.record_dir, reverse)?;
let file = Self::find(
&self.rows.list,
&format!("{}|{lrpage}|{collection}|{cursor}", self.name),
)
.or_else(|| {
Self::find(
&self.rows.list_one,
&format!("{}|{lrdir}|{collection}|{cursor}", self.name),
)
})?;
Some(serde_json::from_slice(&std::fs::read(self.root.join(file)).unwrap()).unwrap())
}
}
fn rkeys(page: &Value) -> Vec<String> {
page["records"]
.as_array()
.unwrap()
.iter()
.map(|record| {
record["uri"]
.as_str()
.unwrap()
.rsplit('/')
.next()
.unwrap()
.into()
})
.collect()
}
#[test]
fn every_limit_and_direction_chains_through_every_record_once() {
let repo = repo(230, 1);
let emitted = emitted(&repo);
for limit in [1, 3, 10, 49, 50, 99, 100] {
for (reverse, is_reverse) in [("", false), ("false", false), ("true", true)] {
let mut seen = Vec::new();
let mut cursor = String::new();
loop {
let page = emitted
.list_records(&limit.to_string(), reverse, COLLECTION, &cursor)
.unwrap_or_else(|| {
panic!("missing page at limit={limit} reverse={reverse} cursor={cursor}")
});
let keys = rkeys(&page);
assert!(
!keys.is_empty() && keys.len() <= limit,
"limit={limit} got {}",
keys.len()
);
seen.extend(keys.clone());
match page["cursor"].as_str() {
Some(next) => {
assert_eq!(Some(next), keys.last().map(String::as_str));
cursor = next.to_string();
}
None => break,
}
}
let mut want: Vec<String> = (0..230).rev().map(|n| format!("r{n:03}")).collect();
if is_reverse {
want.reverse();
}
assert_eq!(seen, want, "limit={limit} reverse={reverse}");
}
}
}
#[test]
fn switching_limit_or_starting_from_an_off_grid_rkey_chains_to_the_end() {
let repo = repo(230, 1);
let emitted = emitted(&repo);
for (reverse, is_reverse) in [("false", false), ("true", true)] {
let mut want: Vec<String> = (0..230).rev().map(|n| format!("r{n:03}")).collect();
if is_reverse {
want.reverse();
}
let schedule = [50, 3, 100, 49, 10, 1, 99];
let mut seen = Vec::new();
let mut cursor = String::new();
for step in 0.. {
let limit = schedule[step % schedule.len()];
let page = emitted
.list_records(&limit.to_string(), reverse, COLLECTION, &cursor)
.unwrap();
let keys = rkeys(&page);
assert!(keys.len() <= limit);
seen.extend(keys);
match page["cursor"].as_str() {
Some(next) => cursor = next.to_string(),
None => break,
}
}
assert_eq!(seen, want, "limit switch reverse={reverse}");
let start_cursor = &want[17];
let mut from_cursor = Vec::new();
let mut cursor = start_cursor.clone();
loop {
let page = emitted
.list_records("50", reverse, COLLECTION, &cursor)
.unwrap();
let keys = rkeys(&page);
assert!(keys.len() <= 50);
from_cursor.extend(keys);
match page["cursor"].as_str() {
Some(next) => cursor = next.to_string(),
None => break,
}
}
assert_eq!(from_cursor, want[18..], "off-grid cursor reverse={reverse}");
let last_rkey = want.last().unwrap();
let after_last = emitted
.list_records("50", reverse, COLLECTION, last_rkey)
.unwrap();
assert!(rkeys(&after_last).is_empty());
assert!(after_last["cursor"].is_null());
}
}
#[test]
fn a_last_page_has_a_null_cursor_a_full_one_does_not_and_cursors_are_exclusive() {
let repo = repo(100, 1);
let name = repo.dir_name();
let emitted = emitted(&repo);
let key = |size: usize, cursor: &str| format!("{name}|{size}f|{COLLECTION}|{cursor}");
let only = emitted.page(&emitted.rows.list, &key(100, ""));
assert_eq!(only["records"].as_array().unwrap().len(), 100);
assert!(only.get("cursor").is_some_and(Value::is_null), "{only}");
let second = emitted.page(&emitted.rows.list, &key(50, "r050"));
assert_eq!(rkeys(&second).first().map(String::as_str), Some("r049"));
assert!(second["cursor"].is_null());
}
#[test]
fn a_record_in_a_list_page_is_its_get_record_answer() {
let repo = repo(3, 1);
let name = repo.dir_name();
let emitted = emitted(&repo);
let page = emitted.page(&emitted.rows.list, &format!("{name}|50f|{COLLECTION}|"));
let answers: Vec<&Value> = repo
.records
.iter()
.rev()
.map(|record| &record.answer)
.collect();
assert_eq!(
page["records"]
.as_array()
.unwrap()
.iter()
.collect::<Vec<_>>(),
answers
);
}
#[test]
fn pages_with_the_same_records_are_one_file() {
let repo = repo(3, 1);
let name = repo.dir_name();
let emitted = emitted(&repo);
let file = |size: usize| {
let key = format!("{name}|{size}f|{COLLECTION}|");
emitted
.rows
.list
.iter()
.find(|(row, _)| *row == key)
.unwrap()
.1
.clone()
};
assert_eq!(file(10), file(50));
assert_eq!(file(50), file(100));
let files = std::fs::read_dir(emitted.root.join(&name).join("lists"))
.unwrap()
.count();
assert_eq!(
files,
5 + 2,
"size 1 has 5 distinct pages across both directions (the middle page r001+cursor=r001 is shared), and sizes 10/50/100 share one full page per direction"
);
}
#[test]
fn a_list_page_is_named_by_the_bytes_its_json_map_writes() {
let mut repo = repo(120, 1);
for (n, record) in repo.records.iter_mut().enumerate() {
record.answer["value"] = json!({
"text": format!("é \"quoted\" \\ \u{1}{n} 🐱"),
"nested": { "z": [n, null, true, 1.5], "a": "" },
});
}
let name = repo.dir_name();
let emitted = emitted(&repo);
let dir = emitted.root.join(&name).join("lists");
let mut pages = 0;
for entry in std::fs::read_dir(&dir).unwrap() {
let path = entry.unwrap().path();
let bytes = std::fs::read(&path).unwrap();
let page: Value = serde_json::from_slice(&bytes).unwrap();
let written = serde_json::to_vec(&json!({
"records": page["records"],
"cursor": page["cursor"],
}))
.unwrap();
assert!(bytes == written, "{}", path.display());
assert_eq!(
path.file_stem().unwrap().to_str().unwrap(),
short_hash(&bytes)
);
pages += 1;
}
assert!(pages > 100, "{pages}");
}
#[test]
fn cursors_are_spelled_both_ways_and_unknown_cursors_miss() {
let mut repo = repo(60, 1);
repo.records[50].rkey = "r050:b".into();
let name = repo.dir_name();
let emitted = emitted(&repo);
for cursor in ["r050:b", "r050%3Ab"] {
let key = format!("{name}|50f|{COLLECTION}|{cursor}");
assert!(
emitted.rows.list.iter().any(|(row, _)| *row == key),
"{key}"
);
}
assert!(
emitted
.list_records("50", "false", COLLECTION, "nonexistent")
.is_none()
);
}
#[test]
fn adding_a_newer_record_keeps_every_full_older_page() {
let before = emitted(&repo(230, 1));
let after = emitted(&repo(231, 1));
let list_files = |e: &Emitted| -> std::collections::BTreeSet<String> {
std::fs::read_dir(e.root.join(&e.name).join("lists"))
.unwrap()
.map(|entry| entry.unwrap().file_name().into_string().unwrap())
.collect()
};
let before_files = list_files(&before);
let after_files = list_files(&after);
let added = after_files.difference(&before_files).count();
assert_eq!(
added, 5,
"1 new record adds only the new size-1 page (f), the updated prior last size-1 page (r), and the new 1-record head/tail pages for sizes 10, 50, 100"
);
}
#[test]
fn get_record_rows_name_the_current_cid_and_no_other() {
let repo = repo(2, 1);
let name = repo.dir_name();
let emitted = emitted(&repo);
let cid = repo.records[1].answer["cid"].as_str().unwrap();
let rows: Vec<&str> = emitted
.rows
.record
.iter()
.filter(|(_, file)| *file == record_file(COLLECTION, "r001", cid))
.map(|(key, _)| key.as_str())
.collect();
assert_eq!(
rows,
[
format!("{name}|{COLLECTION}|r001|"),
format!("{name}|{COLLECTION}|r001|{cid}")
]
);
}
#[test]
fn versions_page_by_offset_in_both_orders_under_one_uri() {
let repo = repo(1, 101);
let emitted = emitted(&repo);
let uri = format!("at://{}/{COLLECTION}/r000", repo.did);
let key = |size: usize, label: char, offset: usize| format!("{uri}|{size}{label}|{offset}");
let first = emitted.page(&emitted.rows.version, &key(100, 'a', 0));
assert_eq!(first["uri"], uri.as_str());
assert_eq!(
(first["total"].as_u64(), first["cursor"].as_str()),
(Some(101), Some("100"))
);
assert_eq!(first["versions"].as_array().unwrap().len(), 100);
assert_eq!(first["versions"][0]["changeId"], format!("{:040x}", 0));
let last = emitted.page(&emitted.rows.version, &key(100, 'a', 100));
assert_eq!(last["versions"].as_array().unwrap().len(), 1);
assert!(last.get("cursor").is_none(), "{last}");
let newest = emitted.page(&emitted.rows.version, &key(500, 'd', 0));
assert_eq!(newest["versions"][0]["cid"], format!("bafyrei{:052}", 100));
assert_eq!(newest["versions"][100]["cid"], format!("bafyrei{:052}", 0));
assert!(newest.get("cursor").is_none());
assert!(
!emitted
.rows
.version
.iter()
.any(|(row, _)| *row == key(500, 'a', 100))
);
}
#[test]
fn a_cob_with_one_page_per_size_costs_six_rows() {
let repo = repo(1, 5);
let emitted = emitted(&repo);
let uri = format!("at://{}/{COLLECTION}/r000", repo.did);
assert_eq!(
emitted.rows.version_uri,
[(uri.clone(), uri.clone()), (form_encoded(&uri), uri)]
);
assert_eq!(emitted.rows.version.len(), 4, "2 sizes x 2 orders");
}
#[test]
fn limits_and_directions_map_to_the_pages_that_exist() {
let rows = Rows::paging();
let page = |rows: &[(String, String)], key: &str| {
rows.iter()
.find(|(row, _)| row == key)
.map(|(_, page)| page.clone())
};
assert_eq!(page(&rows.record_page, "|").as_deref(), Some("50f"));
assert_eq!(page(&rows.record_page, "1|false").as_deref(), Some("1f"));
assert_eq!(page(&rows.record_page, "3|").as_deref(), Some("1f"));
assert_eq!(page(&rows.record_page, "9|true").as_deref(), Some("1r"));
assert_eq!(page(&rows.record_page, "10|").as_deref(), Some("10f"));
assert_eq!(page(&rows.record_page, "49|").as_deref(), Some("10f"));
assert_eq!(page(&rows.record_page, "50|").as_deref(), Some("50f"));
assert_eq!(page(&rows.record_page, "99|true").as_deref(), Some("50r"));
assert_eq!(page(&rows.record_page, "100|true").as_deref(), Some("100r"));
for bad in [
"0|", "-1|", "101|", "500|", "1000|", "5000|", "abc|", "50|maybe",
] {
assert_eq!(page(&rows.record_page, bad), None, "{bad}");
}
assert_eq!(page(&rows.version_page, "|desc").as_deref(), Some("100d"));
assert_eq!(page(&rows.version_page, "500|asc").as_deref(), Some("500a"));
assert_eq!(page(&rows.version_page, "500|sideways"), None);
}
#[test]
fn a_missing_version_uri_is_not_found_only_in_a_versioned_collection() {
let rows = version_misses();
let regex = &rows.last().unwrap().0;
assert!(regex.starts_with("~(^|/|%2F)(") && regex.contains(r"org\.tangled\.track\.ticket"));
assert!(!regex.contains(r"org\.tangled\.git\.refs"));
}