use super::types::{
ChangeId, ChangeLoadBatch, ChangeLoadRequest, ChangeScanBatch, ChangeScanRequest,
ChangelogAppend, CommitId, CommitLoadBatch, CommitLoadRequest, CommitScanBatch,
CommitScanRequest,
};
use crate::common::LixError;
use crate::storage_adapter::{StorageSpace, StorageSpaceId, ValueSemantics};
use async_trait::async_trait;
pub(crate) const COMMIT_NAMESPACE: &str = "changelog.commit";
pub(crate) const CHANGE_NAMESPACE: &str = "changelog.change";
pub(crate) const COMMIT_SPACE: StorageSpace = StorageSpace::declare(
StorageSpaceId(0x0006_0001),
COMMIT_NAMESPACE,
ValueSemantics::Mutable,
);
pub(crate) const CHANGE_SPACE: StorageSpace = StorageSpace::declare(
StorageSpaceId(0x0006_0002),
CHANGE_NAMESPACE,
ValueSemantics::Mutable,
);
pub(crate) fn commit_key(commit_id: CommitId) -> Vec<u8> {
commit_id.as_uuid().as_bytes().to_vec()
}
pub(crate) fn change_key(change_id: ChangeId) -> Vec<u8> {
change_id.as_uuid().as_bytes().to_vec()
}
pub(crate) fn commit_id_from_key(key: &[u8]) -> Result<CommitId, LixError> {
uuid_from_key(key, "commit").map(CommitId::new)
}
pub(crate) fn change_id_from_key(key: &[u8]) -> Result<ChangeId, LixError> {
uuid_from_key(key, "change").map(ChangeId::new)
}
fn uuid_from_key(key: &[u8], kind: &str) -> Result<uuid::Uuid, LixError> {
uuid::Uuid::from_slice(key).map_err(|error| {
LixError::new(
LixError::CODE_INTERNAL_ERROR,
format!("changelog {kind} key is not a 16-byte uuid: {error}"),
)
})
}
#[async_trait]
pub(crate) trait ChangelogReader {
async fn load_commits<'a>(
&mut self,
request: CommitLoadRequest<'a>,
) -> Result<CommitLoadBatch<'a>, LixError>;
async fn scan_commits(
&mut self,
request: CommitScanRequest<'_>,
) -> Result<CommitScanBatch, LixError>;
async fn load_changes<'a>(
&mut self,
request: ChangeLoadRequest<'a>,
) -> Result<ChangeLoadBatch<'a>, LixError>;
async fn scan_changes(
&mut self,
request: ChangeScanRequest<'_>,
) -> Result<ChangeScanBatch, LixError>;
}
#[async_trait]
pub(crate) trait ChangelogWriter {
async fn stage_append(&mut self, append: ChangelogAppend) -> Result<(), LixError>;
async fn stage_delete_standalone_changes(
&mut self,
change_ids: &[ChangeId],
) -> Result<(), LixError>;
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn namespaces_are_stable() {
assert_eq!(COMMIT_NAMESPACE, "changelog.commit");
assert_eq!(CHANGE_NAMESPACE, "changelog.change");
}
#[test]
fn identity_keys_use_raw_uuid_bytes() {
let commit_id = CommitId::for_test_label("commit-1");
let change_id = ChangeId::for_test_label("change-1");
assert_eq!(
commit_key(commit_id),
commit_id.as_uuid().as_bytes().to_vec()
);
assert_eq!(
change_key(change_id),
change_id.as_uuid().as_bytes().to_vec()
);
assert_eq!(commit_key(commit_id).len(), 16);
}
#[test]
fn commit_change_id_is_the_commit_id_at_ordinal_zero() {
let commit_id =
CommitId::with_change_address_space(*CommitId::for_test_label("commit-1").as_uuid());
let change_id = commit_id.commit_change_id();
assert_eq!(change_key(change_id), commit_key(commit_id));
assert_eq!(change_id.as_commit_change(), Some(commit_id));
}
#[test]
fn identity_key_order_matches_text_order() {
let mut ids = (0..32)
.map(|index| CommitId::for_test_label(&format!("commit-{index}")))
.collect::<Vec<_>>();
ids.sort_by_key(|id| commit_key(*id));
let text_sorted = {
let mut text = ids.iter().map(CommitId::to_string).collect::<Vec<_>>();
text.sort();
text
};
assert_eq!(
ids.iter().map(CommitId::to_string).collect::<Vec<_>>(),
text_sorted,
"binary key order must match hyphenated text order"
);
}
#[test]
fn commit_record_write_sites_are_the_sanctioned_ones() {
let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
let mut files = Vec::new();
let mut stack = vec![src.clone()];
while let Some(dir) = stack.pop() {
for entry in std::fs::read_dir(&dir).expect("src dir should read") {
let path = entry.expect("dir entry should read").path();
if path.is_dir() {
stack.push(path);
} else if path.extension().is_some_and(|extension| extension == "rs") {
files.push(path);
}
}
}
files.sort();
let needle = concat!("COMMIT", "_SPACE");
let verbs = [".put(", ".delete(", ".delete_batch(", ".stage("];
let mut found = std::collections::BTreeMap::<String, usize>::new();
for file in &files {
let text = std::fs::read_to_string(file).expect("source file should read");
let relative = file
.strip_prefix(&src)
.expect("scanned file should live under src")
.to_string_lossy()
.replace('\\', "/");
let mut from = 0usize;
while let Some(offset) = text[from..].find(needle) {
let at = from + offset;
from = at + needle.len();
let window = &text[at.saturating_sub(160)..at];
let Some((position, verb)) = verbs
.iter()
.filter_map(|verb| window.rfind(verb).map(|position| (position, *verb)))
.max_by_key(|(position, _)| *position)
else {
continue;
};
if window[position..].contains(';') || window[position..].contains('}') {
continue;
}
*found.entry(format!("{relative} {verb}")).or_default() += 1;
}
}
let expected: std::collections::BTreeMap<String, usize> = [
("changelog/context.rs .stage(", 2),
("changelog/gc.rs .delete(", 1),
("migration/api.rs .put(", 2),
("changelog/gc.rs .delete_batch(", 1),
("commit_graph/walker.rs .delete(", 2),
("commit_graph/walker.rs .put(", 4),
("tracked_state/context.rs .put(", 1),
("tracked_state/storage.rs .put(", 1),
("sync/partial_upload.rs .put(", 1),
("sync/partial_merge_analysis/ancestry_tests.rs .put(", 1),
("migration/incorporation/tests.rs .delete(", 1),
("sync/snapshot_omission_migration_tests.rs .delete(", 1),
]
.into_iter()
.map(|(site, count)| (site.to_string(), count))
.collect();
assert_eq!(
found, expected,
"the set of places that write a commit record changed. If you added \
a writer, the persisted-path-index design's single-choke-point \
premise needs re-deriving -- see this test's doc comment -- and any \
index must be maintained at the new site too, or refuse coverage. \
If you only moved code, update the expectation."
);
}
}