use std::collections::{BTreeSet, HashMap, HashSet};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use sha2::{Digest, Sha256};
use crate::application::vcs::{
AsOfSpec, Author, CheckoutInput, CheckoutTarget, Commit, CommitHash, Conflict,
CreateBranchInput, CreateCommitInput, CreateTagInput, Diff, DiffChange, DiffEntry, DiffInput,
LogInput, LogRange, MergeInput, MergeOutcome, MergeStrategy, Ref, RefKind, RefName, ResetInput,
ResetMode, Status, StatusInput,
};
use crate::application::vcs_collections as vc;
use crate::json::{Map, Value as JsonValue};
use crate::runtime::RedDBRuntime;
use crate::storage::transaction::snapshot::{Snapshot, SnapshotManager, Xid, XID_NONE};
use crate::storage::unified::entity::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
use crate::storage::unified::UnifiedStore;
use crate::{RedDBError, RedDBResult};
use reddb_types::Value;
#[path = "vcs_branches.rs"]
mod branches;
#[path = "vcs_merge_data.rs"]
mod merge_data;
#[cfg(test)]
#[path = "vcs_metadata_tests.rs"]
mod metadata_tests;
impl RedDBRuntime {
pub(crate) fn vcs_live_graph_validation_contexts(
&self,
collection: &str,
) -> RedDBResult<Vec<reddb_engine::transaction::read::SnapshotContext>> {
let (mut context, _retention) = self.native_read_snapshot()?;
context.snapshot = self.snapshot_manager().fresh_read_snapshot();
context.own_xids.clear();
if !self.vcs_is_versioned(collection)? {
return Ok(vec![context]);
}
let store = self.db().store();
let mut names = BTreeSet::from([vc::DEFAULT_BRANCH_REF.to_string()]);
if let Some(manager) = store.get_collection(vc::REFS) {
manager.try_for_each_entity(usize::MAX, |entity| {
if let Some(name) = entity.data.as_row().and_then(|row| row_text(row, "id")) {
if name.starts_with(vc::BRANCH_REF_PREFIX) {
names.insert(name);
}
}
true
})?;
}
names
.into_iter()
.map(|name| {
let mut branch = match branches::load_view(&store, &name)? {
Some(branch) => branch,
None if name == vc::DEFAULT_BRANCH_REF => {
reddb_engine::transaction::read::BranchReadView {
writer_epoch: 0,
epochs: Default::default(),
}
}
None => {
return Err(RedDBError::InvalidOperation(format!(
"checkout legacy branch '{name}' before enabling a graph contract"
)))
}
};
branch
.epochs
.insert(branch.writer_epoch, context.snapshot.clone());
let mut snapshot = context.clone();
snapshot.branch = Some(Arc::new(branch));
Ok(snapshot)
})
.collect()
}
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
fn row_text(row: &RowData, field: &str) -> Option<String> {
match row.get_field(field)?.clone() {
Value::Text(v) => Some(v.to_string()),
_ => None,
}
}
fn row_u64(row: &RowData, field: &str) -> Option<u64> {
match row.get_field(field)?.clone() {
Value::UnsignedInteger(v) => Some(v),
Value::Integer(v) if v >= 0 => Some(v as u64),
_ => None,
}
}
fn row_i64(row: &RowData, field: &str) -> Option<i64> {
match row.get_field(field)?.clone() {
Value::Integer(v) => Some(v),
Value::UnsignedInteger(v) => Some(v as i64),
Value::TimestampMs(v) | Value::Timestamp(v) => Some(v),
_ => None,
}
}
fn row_json(row: &RowData, field: &str) -> JsonValue {
match row.get_field(field) {
Some(Value::Json(bytes)) => {
crate::json::from_slice::<JsonValue>(bytes).unwrap_or(JsonValue::Null)
}
Some(Value::Text(s)) => crate::json::from_str::<JsonValue>(s)
.unwrap_or_else(|_| JsonValue::String(s.to_string())),
_ => JsonValue::Null,
}
}
fn row_string_list(row: &RowData, field: &str) -> Vec<String> {
match row.get_field(field) {
Some(Value::Array(items)) => items
.iter()
.filter_map(|v| match v {
Value::Text(s) => Some(s.to_string()),
_ => None,
})
.collect(),
_ => Vec::new(),
}
}
fn compute_commit_hash(
root_xid: Xid,
parents: &[CommitHash],
author: &Author,
message: &str,
timestamp_ms: i64,
) -> CommitHash {
let mut h = Sha256::new();
h.update(b"reddb-commit-v1\n");
h.update(root_xid.to_be_bytes());
let mut sorted = parents.to_vec();
sorted.sort();
for p in &sorted {
h.update(b"\np=");
h.update(p.as_bytes());
}
h.update(b"\na=");
h.update(author.name.as_bytes());
h.update(b"\n");
h.update(author.email.as_bytes());
h.update(b"\nm=");
h.update(message.as_bytes());
h.update(b"\nt=");
h.update(timestamp_ms.to_be_bytes());
let digest = h.finalize();
hex::encode(digest)
}
fn normalize_branch_name(raw: &str) -> String {
if raw.starts_with(vc::BRANCH_REF_PREFIX) {
raw.to_string()
} else {
format!("{}{}", vc::BRANCH_REF_PREFIX, raw)
}
}
fn normalize_tag_name(raw: &str) -> String {
if raw.starts_with(vc::TAG_REF_PREFIX) {
raw.to_string()
} else {
format!("{}{}", vc::TAG_REF_PREFIX, raw)
}
}
fn head_ref_id(connection_id: u64) -> String {
format!("{}{}", vc::HEAD_ID_PREFIX, connection_id)
}
fn load_metadata_entity(
store: &UnifiedStore,
collection: &str,
key: &str,
) -> RedDBResult<Option<UnifiedEntity>> {
let mut result = None;
if let Some(manager) = store.get_collection(collection) {
super::function_budget::try_scan(
|visit| manager.try_for_each_entity(usize::MAX, visit),
|entity| {
if entity
.data
.as_row()
.is_some_and(|row| row_text(row, "id").as_deref() == Some(key))
{
result = Some(entity.clone());
return false;
}
true
},
)?;
}
Ok(result)
}
fn load_commit_entity(store: &UnifiedStore, hash: &str) -> RedDBResult<Option<UnifiedEntity>> {
load_metadata_entity(store, vc::COMMITS, hash)
}
fn commit_from_row(row: &RowData) -> Option<Commit> {
Some(Commit {
hash: row_text(row, "id")?,
root_xid: row_u64(row, "root_xid")?,
parents: row_string_list(row, "parents"),
height: row_u64(row, "height").unwrap_or(0),
author: Author {
name: row_text(row, "author_name").unwrap_or_default(),
email: row_text(row, "author_email").unwrap_or_default(),
},
committer: Author {
name: row_text(row, "committer_name").unwrap_or_default(),
email: row_text(row, "committer_email").unwrap_or_default(),
},
message: row_text(row, "message").unwrap_or_default(),
timestamp_ms: row_i64(row, "timestamp_ms").unwrap_or(0),
signature: row_text(row, "signature"),
})
}
fn load_commit(store: &UnifiedStore, hash: &str) -> RedDBResult<Option<Commit>> {
load_commit_entity(store, hash)?
.map(|entity| {
entity
.data
.as_row()
.and_then(commit_from_row)
.ok_or_else(|| RedDBError::Internal(format!("vcs: invalid commit '{hash}'")))
})
.transpose()
}
struct CommitSnapshot<'a> {
manager: &'a SnapshotManager,
snapshot: Snapshot,
retained: bool,
}
impl std::ops::Deref for CommitSnapshot<'_> {
type Target = Snapshot;
fn deref(&self) -> &Snapshot {
&self.snapshot
}
}
impl Drop for CommitSnapshot<'_> {
fn drop(&mut self) {
if !self.retained {
self.manager.unpin(
self.snapshot
.in_progress
.iter()
.copied()
.fold(self.snapshot.xid, Xid::min),
);
}
}
}
fn capture_commit_snapshot(manager: &SnapshotManager) -> CommitSnapshot<'_> {
let xid = manager.begin();
let mut snapshot = manager.snapshot(xid);
snapshot.in_progress.retain(|pending| *pending < xid);
let horizon = snapshot.in_progress.iter().copied().fold(xid, Xid::min);
manager.pin(horizon);
manager.commit(xid);
CommitSnapshot {
manager,
snapshot,
retained: false,
}
}
fn commit_snapshot_from_row(row: &RowData) -> RedDBResult<Snapshot> {
let xid = row_u64(row, "root_xid")
.ok_or_else(|| RedDBError::Internal("vcs: commit has no root snapshot".to_string()))?;
let mut in_progress = HashSet::new();
if let Some(value) = row.get_field("snapshot_in_progress") {
let Value::Array(values) = value else {
return Err(RedDBError::Internal(
"vcs: invalid commit snapshot exclusions".to_string(),
));
};
for value in values {
let pending = match value {
Value::UnsignedInteger(value) => *value,
Value::Integer(value) if *value > 0 => *value as u64,
_ => {
return Err(RedDBError::Internal(
"vcs: invalid commit snapshot exclusion".to_string(),
));
}
};
if pending == XID_NONE || pending >= xid {
return Err(RedDBError::Internal(
"vcs: commit snapshot exclusion outside its horizon".to_string(),
));
}
in_progress.insert(pending);
}
}
Ok(Snapshot { xid, in_progress })
}
fn snapshot_sees_entity(
context: &reddb_engine::transaction::read::SnapshotContext,
entity: &UnifiedEntity,
) -> bool {
reddb_engine::transaction::read::entity_visibility_check(context, entity)
}
fn commit_fields(
commit: &Commit,
snapshot: &Snapshot,
view: &reddb_engine::transaction::read::BranchReadView,
) -> HashMap<String, Value> {
let mut fields: HashMap<String, Value> = HashMap::new();
fields.insert("branch_lineage".into(), branches::encode_epochs(view));
fields.insert(
"writer_epoch".into(),
Value::UnsignedInteger(view.writer_epoch),
);
fields.insert("id".to_string(), Value::text(commit.hash.as_str()));
fields.insert(
"root_xid".to_string(),
Value::UnsignedInteger(commit.root_xid),
);
let mut pending: Vec<Xid> = snapshot.in_progress.iter().copied().collect();
pending.sort_unstable();
fields.insert(
"snapshot_in_progress".to_string(),
Value::Array(pending.into_iter().map(Value::UnsignedInteger).collect()),
);
fields.insert(
"parents".to_string(),
Value::Array(
commit
.parents
.iter()
.map(|p| Value::text(p.as_str()))
.collect(),
),
);
fields.insert("height".to_string(), Value::UnsignedInteger(commit.height));
fields.insert(
"author_name".to_string(),
Value::text(commit.author.name.as_str()),
);
fields.insert(
"author_email".to_string(),
Value::text(commit.author.email.as_str()),
);
fields.insert(
"committer_name".to_string(),
Value::text(commit.committer.name.as_str()),
);
fields.insert(
"committer_email".to_string(),
Value::text(commit.committer.email.as_str()),
);
fields.insert("message".to_string(), Value::text(commit.message.as_str()));
fields.insert(
"timestamp_ms".to_string(),
Value::TimestampMs(commit.timestamp_ms),
);
if let Some(sig) = &commit.signature {
fields.insert("signature".to_string(), Value::text(sig.as_str()));
}
fields
}
fn ref_kind_from_str(s: &str) -> Option<RefKind> {
match s {
"branch" => Some(RefKind::Branch),
"tag" => Some(RefKind::Tag),
"head" => Some(RefKind::Head),
_ => None,
}
}
fn ref_from_row(row: &RowData) -> Option<Ref> {
Some(Ref {
name: row_text(row, "id")?,
kind: ref_kind_from_str(&row_text(row, "type")?)?,
target: row_text(row, "target")?,
protected: row
.get_field("protected")
.and_then(|v| match v {
Value::Boolean(b) => Some(*b),
_ => None,
})
.unwrap_or(false),
})
}
fn load_ref_entity(
store: &UnifiedStore,
name: &str,
) -> RedDBResult<Option<(EntityId, UnifiedEntity)>> {
Ok(load_metadata_entity(store, vc::REFS, name)?.map(|entity| (entity.id, entity)))
}
fn load_ref(store: &UnifiedStore, name: &str) -> RedDBResult<Option<Ref>> {
load_ref_entity(store, name)?
.map(|(_, entity)| {
entity
.data
.as_row()
.and_then(ref_from_row)
.ok_or_else(|| RedDBError::Internal(format!("vcs: invalid reference '{name}'")))
})
.transpose()
}
fn stage_ref(
store: &UnifiedStore,
publication: &mut crate::storage::unified::store::RowPublication,
reference: &Ref,
) -> RedDBResult<()> {
let kind = match reference.kind {
RefKind::Branch => "branch",
RefKind::Tag => "tag",
RefKind::Head => "head",
};
branches::stage_metadata(
store,
publication,
vc::REFS,
"id",
&reference.name,
HashMap::from([
("id".into(), Value::text(reference.name.as_str())),
("type".into(), Value::text(kind)),
("target".into(), Value::text(reference.target.as_str())),
("protected".into(), Value::Boolean(reference.protected)),
]),
)
}
fn save_ref(store: &UnifiedStore, reference: &Ref) -> RedDBResult<()> {
let mut publication = crate::storage::unified::store::RowPublication::default();
stage_ref(store, &mut publication, reference)?;
store.publish_rows(publication)?;
Ok(())
}
fn delete_ref(store: &UnifiedStore, name: &str) -> RedDBResult<bool> {
let Some((id, _)) = load_ref_entity(store, name)? else {
return Ok(false);
};
let mut publication = crate::storage::unified::store::RowPublication::default();
publication.deletes.push((vc::REFS.into(), id));
store.publish_rows(publication)?;
Ok(true)
}
fn is_versioned(store: &UnifiedStore, name: &str) -> RedDBResult<bool> {
if name.starts_with("red_") || name.starts_with("red.") {
return Ok(false);
}
let Some(manager) = store.get_collection(vc::SETTINGS) else {
return Ok(false);
};
let _publication = manager
.has_backing()
.then(|| store.vault_publication_lock.lock());
let mut versioned = false;
super::function_budget::try_scan(
|visit| manager.try_for_each_entity(usize::MAX, visit),
|entity| {
versioned = entity.data.as_row().is_some_and(|row| {
row.get_field("id")
.is_some_and(|id| matches!(id, Value::Text(id) if id.as_ref() == name))
&& matches!(row.get_field("versioned"), Some(Value::Boolean(true)))
});
!versioned
},
)?;
Ok(versioned)
}
fn versioned_collections(store: &UnifiedStore) -> RedDBResult<Vec<String>> {
let Some(manager) = store.get_collection(vc::SETTINGS) else {
return Ok(Vec::new());
};
let _publication = manager
.has_backing()
.then(|| store.vault_publication_lock.lock());
let mut collections = Vec::new();
super::function_budget::try_scan(
|visit| manager.try_for_each_entity(usize::MAX, visit),
|entity| {
if let Some(row) = entity.data.as_row() {
if matches!(row.get_field("versioned"), Some(Value::Boolean(true))) {
if let Some(name) = row_text(row, "id") {
collections.push(name);
}
}
}
true
},
)?;
Ok(collections)
}
fn list_refs_by_prefix(store: &UnifiedStore, prefix: Option<&str>) -> RedDBResult<Vec<Ref>> {
let mut refs = Vec::new();
if let Some(manager) = store.get_collection(vc::REFS) {
super::function_budget::try_scan_result(
|visit| manager.try_for_each_entity(usize::MAX, visit),
|entity| {
let row = entity
.data
.as_row()
.ok_or_else(|| RedDBError::Internal("vcs: invalid reference row".into()))?;
let reference = ref_from_row(row)
.ok_or_else(|| RedDBError::Internal("vcs: invalid reference".into()))?;
if prefix.is_none_or(|prefix| reference.name.starts_with(prefix)) {
refs.push(reference);
}
Ok(true)
},
)?;
}
Ok(refs)
}
fn ancestor_set(
store: &UnifiedStore,
start: &str,
max_steps: usize,
) -> RedDBResult<HashSet<CommitHash>> {
let mut visited: HashSet<CommitHash> = HashSet::new();
let mut stack: Vec<CommitHash> = vec![start.to_string()];
let mut steps = 0usize;
while let Some(hash) = stack.pop() {
if !visited.insert(hash.clone()) {
continue;
}
{
let c = load_commit(store, &hash)?
.ok_or_else(|| RedDBError::Internal(format!("vcs: missing ancestor '{hash}'")))?;
for p in c.parents {
if !visited.contains(&p) {
stack.push(p);
}
}
}
steps += 1;
if steps >= max_steps && !stack.is_empty() {
return Err(RedDBError::InvalidOperation(
"vcs: ancestry exceeds traversal budget".into(),
));
}
}
Ok(visited)
}
fn topo_walk(store: &UnifiedStore, start: &str, range: &LogRange) -> RedDBResult<Vec<Commit>> {
let limit = range.limit.unwrap_or(usize::MAX);
let skip = range.skip.unwrap_or(0);
let exclude = range
.from
.as_ref()
.map(|h| ancestor_set(store, h, 100_000))
.transpose()?
.unwrap_or_default();
let mut visited: HashSet<CommitHash> = HashSet::new();
let mut stack: Vec<CommitHash> = vec![start.to_string()];
let mut out: Vec<Commit> = Vec::new();
let mut skipped = 0usize;
while let Some(hash) = stack.pop() {
if !visited.insert(hash.clone()) {
continue;
}
if exclude.contains(&hash) {
continue;
}
let commit = load_commit(store, &hash)?
.ok_or_else(|| RedDBError::Internal(format!("vcs: missing ancestor '{hash}'")))?;
if range.no_merges && commit.parents.len() > 1 {
for p in &commit.parents {
if !visited.contains(p) {
stack.push(p.clone());
}
}
continue;
}
for p in &commit.parents {
if !visited.contains(p) {
stack.push(p.clone());
}
}
if skipped < skip {
skipped += 1;
continue;
}
if out.len() >= limit {
break;
}
out.push(commit);
}
out.sort_by(|a, b| {
b.height
.cmp(&a.height)
.then(b.timestamp_ms.cmp(&a.timestamp_ms))
});
Ok(out)
}
fn resolve_ref_chain(store: &UnifiedStore, name: &str) -> RedDBResult<Option<CommitHash>> {
let mut current = name.to_string();
for _ in 0..4 {
let Some(reference) = load_ref(store, ¤t)? else {
return Ok(None);
};
match reference.kind {
RefKind::Branch | RefKind::Tag => return Ok(Some(reference.target)),
RefKind::Head => current = reference.target,
}
}
Err(RedDBError::InvalidOperation(
"vcs: reference chain exceeds four levels".into(),
))
}
fn resolve_short_commit(store: &UnifiedStore, prefix: &str) -> RedDBResult<Option<CommitHash>> {
if prefix.len() < 4 {
return Ok(None);
}
let mut found = None;
let mut ambiguous = false;
if let Some(manager) = store.get_collection(vc::COMMITS) {
super::function_budget::try_scan(
|visit| manager.try_for_each_entity(usize::MAX, visit),
|entity| {
if let Some(hash) = entity.data.as_row().and_then(|row| row_text(row, "id")) {
if hash.starts_with(prefix) {
ambiguous = found.is_some();
found = Some(hash);
}
}
!ambiguous
},
)?;
}
Ok(if ambiguous { None } else { found })
}
fn stage_workset(
store: &UnifiedStore,
publication: &mut crate::storage::unified::store::RowPublication,
connection_id: u64,
branch: &str,
base_commit: Option<&str>,
working_xid: Xid,
) -> RedDBResult<()> {
let connection = connection_id.to_string();
let mut fields = HashMap::from([
("id".into(), Value::text(connection.as_str())),
("branch".into(), Value::text(branch)),
("working_xid".into(), Value::UnsignedInteger(working_xid)),
]);
if let Some(base) = base_commit {
fields.insert("base_commit".into(), Value::text(base));
}
branches::stage_metadata(store, publication, vc::WORKSETS, "id", &connection, fields)
}
fn load_workset(
store: &UnifiedStore,
connection_id: u64,
) -> RedDBResult<Option<(RefName, Option<CommitHash>)>> {
Ok(load_workset_state(store, connection_id)?.map(|(branch, base, _)| (branch, base)))
}
fn load_workset_state(
store: &UnifiedStore,
connection_id: u64,
) -> RedDBResult<Option<(RefName, Option<CommitHash>, Option<String>)>> {
let Some(entity) = load_metadata_entity(store, vc::WORKSETS, &connection_id.to_string())?
else {
return Ok(None);
};
let row = entity
.data
.as_row()
.ok_or_else(|| RedDBError::Internal("vcs: invalid workset".into()))?;
let branch = row_text(row, "branch")
.ok_or_else(|| RedDBError::Internal("vcs: workset has no branch".into()))?;
let base = row_text(row, "base_commit");
let pending_merge = row_text(row, "merge_state_id");
let head = if branch.starts_with(vc::BRANCH_REF_PREFIX) && pending_merge.is_none() {
load_ref(store, &branch)?
.map(|reference| reference.target)
.or(base)
} else {
base
};
Ok(Some((branch, head, pending_merge)))
}
fn ensure_no_pending_merge(store: &UnifiedStore, connection_id: u64) -> RedDBResult<()> {
let branch = load_workset(store, connection_id)?
.map(|(branch, _)| branch)
.unwrap_or_else(|| vc::DEFAULT_BRANCH_REF.into());
let mut pending = None;
if let Some(manager) = store.get_collection(vc::WORKSETS) {
manager.try_for_each_entity(usize::MAX, |entity| {
if let Some(row) = entity
.data
.as_row()
.filter(|row| row_text(row, "branch").as_deref() == Some(&branch))
{
pending = row_text(row, "merge_state_id");
}
pending.is_none()
})?;
}
if let Some(state) = pending {
return Err(RedDBError::InvalidOperation(format!(
"cannot change branch data or HEAD while merge `{state}` is pending"
)));
}
Ok(())
}
impl RedDBRuntime {
pub(crate) fn vcs_has_versioned_collections(&self) -> RedDBResult<bool> {
Ok(!versioned_collections(&self.db().store())?.is_empty())
}
pub(crate) fn vcs_reject_destructive_ddl(&self, collection: &str) -> RedDBResult<()> {
if self.vcs_is_versioned(collection)? {
return Err(RedDBError::InvalidOperation(format!(
"cannot physically drop or truncate versioned collection '{collection}'; use versioned deletes or QUEUE PURGE"
)));
}
Ok(())
}
pub(super) fn rehydrate_vcs_snapshot_pins(&self) -> RedDBResult<()> {
let store = self.inner.db.store();
let Some(commits) = store.get_collection(vc::COMMITS) else {
return Ok(());
};
let manager = self.inner.transaction_state.snapshot_manager();
let mut failure = None;
commits.try_for_each_entity(usize::MAX, |entity| {
let Some(row) = entity.data.as_row() else {
failure = Some(RedDBError::Internal(
"vcs: invalid commit record".to_string(),
));
return false;
};
match commit_snapshot_from_row(row) {
Ok(snapshot) => {
manager.observe_committed_xid(snapshot.xid);
let horizon = snapshot
.in_progress
.iter()
.copied()
.fold(snapshot.xid, Xid::min);
manager.pin(horizon);
true
}
Err(error) => {
failure = Some(error);
false
}
}
})?;
match failure {
Some(error) => Err(error),
None => Ok(()),
}
}
pub fn vcs_commit(&self, input: CreateCommitInput) -> RedDBResult<Commit> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
if let Some((_, _, Some(state))) = load_workset_state(store, input.connection_id)? {
return merge_data::finish_pending(self, input, &state);
}
ensure_no_pending_merge(store, input.connection_id)?;
let workset = load_workset(store, input.connection_id)?;
let branch_ref = workset
.as_ref()
.map(|(b, _)| b.clone())
.unwrap_or_else(|| vc::DEFAULT_BRANCH_REF.to_string());
if !branch_ref.starts_with(vc::BRANCH_REF_PREFIX) {
return Err(RedDBError::InvalidOperation(
"cannot commit a detached or tag checkout".into(),
));
}
let parent_hash = load_ref(store, &branch_ref)?
.map(|reference| reference.target)
.or_else(|| workset.as_ref().and_then(|(_, base)| base.clone()));
let parents: Vec<CommitHash> = match parent_hash.clone() {
Some(h) if !h.is_empty() => vec![h],
_ => Vec::new(),
};
let mut parent_height = 0;
for parent in &parents {
let commit = load_commit(store, parent)?
.ok_or_else(|| RedDBError::Internal("vcs: missing parent commit".into()))?;
parent_height = parent_height.max(commit.height);
}
let height = if parents.is_empty() {
0
} else {
parent_height + 1
};
let snapshot_manager = self.inner.transaction_state.snapshot_manager();
let mut snapshot = capture_commit_snapshot(&snapshot_manager);
let root_xid = snapshot.xid;
let timestamp_ms = now_ms();
let committer = input.committer.unwrap_or_else(|| input.author.clone());
let hash = compute_commit_hash(
root_xid,
&parents,
&input.author,
&input.message,
timestamp_ms,
);
if load_commit(store, &hash)?.is_some() {
return Err(RedDBError::InvalidConfig(format!(
"commit {hash} already exists (duplicate timestamp+content)"
)));
}
let commit = Commit {
hash: hash.clone(),
root_xid,
parents,
height,
author: input.author,
committer,
message: input.message,
timestamp_ms,
signature: None,
};
if !branch_ref.starts_with(vc::BRANCH_REF_PREFIX) {
return Err(RedDBError::InvalidOperation(
"cannot commit a detached or tag checkout".into(),
));
}
merge_data::publish_commit(
self,
input.connection_id,
&branch_ref,
&commit,
&snapshot,
None,
)?;
snapshot.retained = true;
Ok(commit)
}
pub fn vcs_branch_create(&self, input: CreateBranchInput) -> RedDBResult<Ref> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
let full_name = normalize_branch_name(&input.name);
if load_ref(store, &full_name)?.is_some() {
return Err(RedDBError::InvalidConfig(format!(
"branch `{full_name}` already exists"
)));
}
let target_hash = match &input.from {
Some(spec) => RedDBRuntime::vcs_resolve_commitish(self, spec)?,
None => {
let workset = load_workset(store, input.connection_id)?;
workset
.and_then(|(_, base)| base)
.or(load_ref(store, vc::DEFAULT_BRANCH_REF)?.map(|r| r.target))
.unwrap_or_default()
}
};
let r = Ref {
name: full_name,
kind: RefKind::Branch,
target: target_hash,
protected: false,
};
let mut publication = crate::storage::unified::store::RowPublication::default();
branches::stage_new_view(store, &mut publication, &r.name, &r.target)?;
stage_ref(store, &mut publication, &r)?;
store.publish_rows(publication)?;
Ok(r)
}
pub fn vcs_branch_delete(&self, name: &str) -> RedDBResult<()> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
let full = normalize_branch_name(name);
let existing = load_ref(store, &full)?
.ok_or_else(|| RedDBError::NotFound(format!("branch `{full}` does not exist")))?;
if existing.protected {
return Err(RedDBError::ReadOnly(format!(
"branch `{full}` is protected"
)));
}
if full == vc::DEFAULT_BRANCH_REF {
return Err(RedDBError::InvalidOperation(
"cannot delete the default branch".into(),
));
}
let mut publication = crate::storage::unified::store::RowPublication::default();
if let Some((id, _)) = load_ref_entity(store, &full)? {
publication.deletes.push((vc::REFS.into(), id));
}
branches::stage_delete_view(store, &mut publication, &full)?;
store.publish_rows(publication)?;
Ok(())
}
pub fn vcs_tag_create(&self, input: CreateTagInput) -> RedDBResult<Ref> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
let full_name = normalize_tag_name(&input.name);
if load_ref(store, &full_name)?.is_some() {
return Err(RedDBError::InvalidConfig(format!(
"tag `{full_name}` already exists"
)));
}
let target_hash = RedDBRuntime::vcs_resolve_commitish(self, &input.target)?;
let r = Ref {
name: full_name,
kind: RefKind::Tag,
target: target_hash,
protected: false,
};
save_ref(store, &r)?;
Ok(r)
}
pub fn vcs_tag_delete(&self, name: &str) -> RedDBResult<()> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
let full = normalize_tag_name(name);
let existing = load_ref(store, &full)?
.ok_or_else(|| RedDBError::NotFound(format!("tag `{full}` does not exist")))?;
if existing.protected {
return Err(RedDBError::ReadOnly(format!("tag `{full}` is protected")));
}
delete_ref(store, &full)?;
Ok(())
}
pub fn vcs_list_refs(&self, prefix: Option<&str>) -> RedDBResult<Vec<Ref>> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
Ok(list_refs_by_prefix(store, prefix)?)
}
pub fn vcs_checkout(&self, input: CheckoutInput) -> RedDBResult<Ref> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
if self
.inner
.transaction_state
.context(input.connection_id)
.is_some()
{
return Err(RedDBError::InvalidOperation(
"cannot checkout during an active transaction".into(),
));
}
if !input.force {
ensure_no_pending_merge(store, input.connection_id)?;
}
let mut publication = crate::storage::unified::store::RowPublication::default();
let (branch_ref, target_hash) = match &input.target {
CheckoutTarget::Branch(name) => {
let full = normalize_branch_name(name);
let r = load_ref(store, &full)?.ok_or_else(|| {
RedDBError::NotFound(format!("branch `{full}` does not exist"))
})?;
if full != vc::DEFAULT_BRANCH_REF && branches::load_view(store, &full)?.is_none() {
branches::stage_new_view(store, &mut publication, &full, &r.target)?;
}
(full, r.target)
}
CheckoutTarget::Tag(name) => {
let full = normalize_tag_name(name);
let r = load_ref(store, &full)?
.ok_or_else(|| RedDBError::NotFound(format!("tag `{full}` does not exist")))?;
(full, r.target)
}
CheckoutTarget::Commit(hash) => {
let resolved = RedDBRuntime::vcs_resolve_commitish(self, hash)?;
(String::new(), resolved)
}
};
stage_workset(
store,
&mut publication,
input.connection_id,
&branch_ref,
Some(&target_hash),
self.inner
.transaction_state
.snapshot_manager()
.peek_next_xid(),
)?;
stage_ref(
store,
&mut publication,
&Ref {
name: head_ref_id(input.connection_id),
kind: RefKind::Head,
target: if branch_ref.is_empty() {
target_hash.clone()
} else {
branch_ref.clone()
},
protected: false,
},
)?;
store.publish_rows(publication)?;
Ok(Ref {
name: if branch_ref.is_empty() {
format!("detached:{target_hash}")
} else {
branch_ref
},
kind: if target_hash.is_empty() {
RefKind::Head
} else {
RefKind::Branch
},
target: target_hash,
protected: false,
})
}
pub fn vcs_merge(&self, input: MergeInput) -> RedDBResult<MergeOutcome> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
ensure_no_pending_merge(store, input.connection_id)?;
let from_hash = RedDBRuntime::vcs_resolve_commitish(self, &input.from)?;
let from_commit = load_commit(store, &from_hash)?.ok_or_else(|| {
RedDBError::NotFound(format!("source commit `{from_hash}` not found"))
})?;
let workset = load_workset(store, input.connection_id)?;
let (head_branch, head_hash) = match workset {
Some((branch, Some(head))) => (branch, head),
Some((branch, None)) => {
let head = load_ref(store, &branch)?
.map(|r| r.target)
.unwrap_or_default();
(branch, head)
}
None => {
let head = load_ref(store, vc::DEFAULT_BRANCH_REF)?
.map(|r| r.target)
.unwrap_or_default();
(vc::DEFAULT_BRANCH_REF.to_string(), head)
}
};
if head_hash.is_empty() {
return Err(RedDBError::InvalidConfig(
"cannot merge: HEAD has no commits".to_string(),
));
}
let from_ancestors = ancestor_set(store, &from_hash, 100_000)?;
let can_fast_forward = from_ancestors.contains(&head_hash);
match input.opts.strategy {
MergeStrategy::FastForwardOnly if !can_fast_forward => {
return Err(RedDBError::InvalidConfig(
"not a fast-forward — use --strategy auto or no-ff".to_string(),
));
}
_ => {}
}
if can_fast_forward && input.opts.strategy != MergeStrategy::NoFastForward {
if head_branch.is_empty() {
return Err(RedDBError::InvalidConfig(
"cannot fast-forward a detached HEAD".to_string(),
));
}
branches::ensure_entity_history_supported(self)?;
branches::ensure_clean(self, input.connection_id, &head_hash)?;
merge_data::plan(self, &head_hash, &head_hash, &from_hash, "head-change")?;
branches::publish_head(
store,
input.connection_id,
&head_branch,
&from_commit,
branches::commit_view(store, &from_hash)?,
)?;
return Ok(MergeOutcome {
merge_commit: Some(from_commit),
fast_forward: true,
conflicts: Vec::new(),
merge_state_id: None,
});
}
if !head_branch.starts_with(vc::BRANCH_REF_PREFIX) {
return Err(RedDBError::InvalidOperation(
"cannot merge into a detached or tag checkout".into(),
));
}
branches::ensure_entity_history_supported(self)?;
branches::ensure_clean(self, input.connection_id, &head_hash)?;
let lca = RedDBRuntime::vcs_lca(self, &head_hash, &from_hash)?.ok_or_else(|| {
RedDBError::InvalidOperation("cannot merge unrelated histories".into())
})?;
let message = input
.opts
.message
.unwrap_or_else(|| format!("Merge {} into {}", input.from, head_branch));
let author = input.author.clone();
let timestamp_ms = now_ms();
let parents = vec![head_hash.clone(), from_hash.clone()];
let mut parent_height = 0;
for parent in &parents {
let commit = load_commit(store, parent)?
.ok_or_else(|| RedDBError::Internal("vcs: missing parent commit".into()))?;
parent_height = parent_height.max(commit.height);
}
let height = parent_height + 1;
let snapshot_manager = self.inner.transaction_state.snapshot_manager();
let mut snapshot = capture_commit_snapshot(&snapshot_manager);
let root_xid = snapshot.xid;
let hash = compute_commit_hash(root_xid, &parents, &author, &message, timestamp_ms);
let merge_commit = Commit {
hash: hash.clone(),
root_xid,
parents,
height,
author: author.clone(),
committer: author,
message,
timestamp_ms,
signature: None,
};
let merge_state_id = format!("ms:{}", &hash[..16]);
let plan = merge_data::plan(self, &lca, &head_hash, &from_hash, &merge_state_id)?;
let conflicts = plan.conflicts.clone();
if conflicts.is_empty() {
let result = merge_data::publish_commit(
self,
input.connection_id,
&head_branch,
&merge_commit,
&snapshot,
Some(plan),
);
result?;
snapshot.retained = true;
return Ok(MergeOutcome {
merge_commit: Some(merge_commit),
fast_forward: false,
conflicts,
merge_state_id: None,
});
}
let mut publication = crate::storage::unified::store::RowPublication::default();
stage_merge_conflicts(store, &mut publication, &conflicts)?;
let mut ms_fields: HashMap<String, Value> = HashMap::new();
ms_fields.insert("id".to_string(), Value::text(merge_state_id.as_str()));
ms_fields.insert("kind".to_string(), Value::text("merge"));
ms_fields.insert("branch".to_string(), Value::text(head_branch.as_str()));
ms_fields.insert("base".to_string(), Value::text(lca.as_str()));
ms_fields.insert("ours".to_string(), Value::text(head_hash.as_str()));
ms_fields.insert("theirs".to_string(), Value::text(from_hash.as_str()));
ms_fields.insert(
"conflicts_count".to_string(),
Value::UnsignedInteger(conflicts.len() as u64),
);
branches::stage_metadata(
store,
&mut publication,
vc::MERGE_STATE,
"id",
&merge_state_id,
ms_fields,
)?;
let head_xid = self.vcs_resolve_as_of(AsOfSpec::Commit(head_hash.clone()))?;
let conn_id = input.connection_id.to_string();
branches::stage_metadata(
store,
&mut publication,
vc::WORKSETS,
"id",
&conn_id,
HashMap::from([
("id".into(), Value::text(conn_id.as_str())),
("branch".into(), Value::text(head_branch.as_str())),
("base_commit".into(), Value::text(head_hash.as_str())),
("working_xid".into(), Value::UnsignedInteger(head_xid)),
(
"merge_state_id".into(),
Value::text(merge_state_id.as_str()),
),
]),
)?;
store.publish_rows(publication)?;
Ok(MergeOutcome {
merge_commit: None,
fast_forward: false,
conflicts,
merge_state_id: Some(merge_state_id),
})
}
pub fn vcs_cherry_pick(
&self,
connection_id: u64,
commit: &str,
author: Author,
) -> RedDBResult<MergeOutcome> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
ensure_no_pending_merge(store, connection_id)?;
let src_hash = RedDBRuntime::vcs_resolve_commitish(self, commit)?;
let src_commit = load_commit(store, &src_hash)?.ok_or_else(|| {
RedDBError::NotFound(format!("cherry-pick source `{src_hash}` not found"))
})?;
if src_commit.parents.is_empty() {
return Err(RedDBError::InvalidConfig(
"cannot cherry-pick a root commit".to_string(),
));
}
if src_commit.parents.len() > 1 {
return Err(RedDBError::InvalidConfig(
"cannot cherry-pick a merge commit; resolve manually".to_string(),
));
}
let parent_hash = src_commit.parents[0].clone();
let picked = merge_data::apply_delta(
self,
connection_id,
&parent_hash,
&src_hash,
author,
format!("cherry-pick: {}", src_commit.message),
)?;
Ok(MergeOutcome {
merge_commit: Some(picked),
fast_forward: false,
conflicts: Vec::new(),
merge_state_id: None,
})
}
pub fn vcs_revert(
&self,
connection_id: u64,
commit: &str,
author: Author,
) -> RedDBResult<Commit> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
ensure_no_pending_merge(store, connection_id)?;
let src_hash = RedDBRuntime::vcs_resolve_commitish(self, commit)?;
let src_commit = load_commit(store, &src_hash)?
.ok_or_else(|| RedDBError::NotFound(format!("revert source `{src_hash}` not found")))?;
if src_commit.parents.is_empty() {
return Err(RedDBError::InvalidConfig(
"cannot revert a root commit".to_string(),
));
}
let parent_hash = src_commit.parents[0].clone();
if src_commit.parents.len() != 1 {
return Err(RedDBError::InvalidOperation(
"cannot revert a merge commit without selecting a parent".into(),
));
}
merge_data::apply_delta(
self,
connection_id,
&src_hash,
&parent_hash,
author,
format!("Revert \"{}\"", src_commit.message),
)
}
pub fn vcs_reset(&self, input: ResetInput) -> RedDBResult<()> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
ensure_no_pending_merge(store, input.connection_id)?;
let target_hash = RedDBRuntime::vcs_resolve_commitish(self, &input.target)?;
let target_commit = load_commit(store, &target_hash)?.ok_or_else(|| {
RedDBError::NotFound(format!("target commit `{target_hash}` not found"))
})?;
let workset = load_workset(store, input.connection_id)?;
let branch = workset
.as_ref()
.map(|(b, _)| b.clone())
.unwrap_or_else(|| vc::DEFAULT_BRANCH_REF.to_string());
if !branch.starts_with(vc::BRANCH_REF_PREFIX) {
return Err(RedDBError::InvalidOperation(
"reset requires a branch".into(),
));
}
if self.snapshot_manager().oldest_active_xid().is_some() {
return Err(RedDBError::InvalidOperation(
"finish active transactions before reset".into(),
));
}
match input.mode {
ResetMode::Soft | ResetMode::Mixed => {
let mut publication = crate::storage::unified::store::RowPublication::default();
branches::stage_head(
store,
&mut publication,
input.connection_id,
&branch,
&target_commit,
None,
)?;
store.publish_rows(publication)?;
Ok(())
}
ResetMode::Hard => {
branches::ensure_entity_history_supported(self)?;
merge_data::validate_head_change(self, input.connection_id, &target_hash)?;
branches::publish_head(
store,
input.connection_id,
&branch,
&target_commit,
branches::commit_view(store, &target_hash)?,
)
}
}
}
pub fn vcs_log(&self, input: LogInput) -> RedDBResult<Vec<Commit>> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let start = match &input.range.to {
Some(spec) => RedDBRuntime::vcs_resolve_commitish(self, spec)?,
None => {
let workset = load_workset(store, input.connection_id)?;
workset
.and_then(|(_, base)| base)
.or(load_ref(store, vc::DEFAULT_BRANCH_REF)?.map(|r| r.target))
.unwrap_or_default()
}
};
if start.is_empty() {
return Ok(Vec::new());
}
Ok(topo_walk(store, &start, &input.range)?)
}
pub fn vcs_diff(&self, input: DiffInput) -> RedDBResult<Diff> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let from_hash = RedDBRuntime::vcs_resolve_commitish(self, &input.from)?;
let to_hash = RedDBRuntime::vcs_resolve_commitish(self, &input.to)?;
let from_snap = self.vcs_commit_read_context(&from_hash)?;
let to_snap = self.vcs_commit_read_context(&to_hash)?;
let mut entries: Vec<DiffEntry> = Vec::new();
let mut added = 0usize;
let mut removed = 0usize;
let mut modified = 0usize;
let collections = store.list_collections();
for coll in collections {
if coll.starts_with("red_") || coll.starts_with("red.") {
continue;
}
if !is_versioned(store, &coll)? {
continue;
}
if let Some(filter) = &input.collection {
if filter != &coll {
continue;
}
}
let Some(manager) = store.get_collection(&coll) else {
continue;
};
self.check_collection_privilege(reddb_rql::ast::PolicyAction::Select, &coll)?;
let mut entities = Vec::new();
manager.try_for_each_entity(usize::MAX, |entity| {
entities.push(entity.clone());
true
})?;
for entity in &entities {
let in_from = snapshot_sees_entity(&from_snap, entity);
let in_to = snapshot_sees_entity(&to_snap, entity);
if in_from == in_to {
continue;
}
ensure_vcs_entity_access(
self,
&coll,
entity,
reddb_rql::ast::PolicyAction::Select,
)?;
let entity_id = entity.logical_id().raw().to_string();
let payload = if input.summary_only {
JsonValue::Null
} else {
vcs_entity_body(self, &coll, entity)
};
let change = match (in_from, in_to) {
(false, true) => {
added += 1;
DiffChange::Added { after: payload }
}
(true, false) => {
removed += 1;
DiffChange::Removed { before: payload }
}
_ => unreachable!(),
};
entries.push(DiffEntry {
collection: coll.clone(),
entity_id,
change,
});
}
}
entries = coalesce_modifications(entries, &mut added, &mut removed, &mut modified);
Ok(Diff {
from: from_hash,
to: to_hash,
entries,
added,
removed,
modified,
})
}
pub fn vcs_status(&self, input: StatusInput) -> RedDBResult<Status> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let workset = load_workset_state(store, input.connection_id)?;
let (head_ref, head_commit, merge_state_id) = match workset {
Some((branch, base, merge_state_id)) => {
let base = base.or(load_ref(store, &branch)?.map(|r| r.target));
(Some(branch), base, merge_state_id)
}
None => {
let base = load_ref(store, vc::DEFAULT_BRANCH_REF)?.map(|r| r.target);
(Some(vc::DEFAULT_BRANCH_REF.to_string()), base, None)
}
};
let detached = matches!(&head_ref, Some(s) if s.is_empty());
let unresolved_conflicts = match &merge_state_id {
Some(id) => self.vcs_conflicts_list(id)?.len(),
None => 0,
};
Ok(Status {
connection_id: input.connection_id,
head_ref: head_ref.filter(|s| !s.is_empty()),
head_commit,
detached,
staged_changes: 0,
working_changes: 0,
unresolved_conflicts,
merge_state_id,
})
}
pub fn vcs_lca(&self, a: &str, b: &str) -> RedDBResult<Option<CommitHash>> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let a_hash = RedDBRuntime::vcs_resolve_commitish(self, a)?;
let b_hash = RedDBRuntime::vcs_resolve_commitish(self, b)?;
let a_ancestors = ancestor_set(store, &a_hash, 100_000)?;
let mut visited: HashSet<CommitHash> = HashSet::new();
let mut stack: Vec<CommitHash> = vec![b_hash];
let mut best: Option<(u64, CommitHash)> = None;
while let Some(hash) = stack.pop() {
if !visited.insert(hash.clone()) {
continue;
}
if a_ancestors.contains(&hash) {
let height = load_commit(store, &hash)?.map(|c| c.height).unwrap_or(0);
match &best {
Some((h, _)) if *h >= height => {}
_ => best = Some((height, hash.clone())),
}
continue;
}
{
let commit = load_commit(store, &hash)?.ok_or_else(|| {
RedDBError::Internal(format!("vcs: missing ancestor '{hash}'"))
})?;
for p in commit.parents {
if !visited.contains(&p) {
stack.push(p);
}
}
}
}
Ok(best.map(|(_, h)| h))
}
pub fn vcs_conflicts_list(&self, merge_state_id: &str) -> RedDBResult<Vec<Conflict>> {
let store = self.db().store();
let _publication = store.vault_publication_lock.lock();
merge_data::read_conflicts(self, merge_state_id)
}
pub fn vcs_conflict_resolve(&self, conflict_id: &str, resolved: JsonValue) -> RedDBResult<()> {
let store = self.db().store();
let _publication = store.vault_publication_lock.lock();
let row = merge_data::metadata_row(&store, vc::CONFLICTS, conflict_id)?
.ok_or_else(|| RedDBError::NotFound(format!("conflict `{conflict_id}` not found")))?;
let choice = match &resolved {
JsonValue::String(choice)
if matches!(choice.as_str(), "ours" | "theirs" | "base" | "delete") =>
{
choice
}
_ => {
return Err(RedDBError::InvalidOperation(
"resolve with ours, theirs, base or delete; conflict retained".into(),
))
}
};
let connection = crate::runtime::impl_core::current_connection_id();
let active_state = load_workset_state(&store, connection)?.and_then(|(_, _, state)| state);
if active_state != row_text(&row, "merge_state_id") {
return Err(RedDBError::InvalidOperation(
"conflict belongs to another pending merge".into(),
));
}
let mut fields: HashMap<_, _> = row
.iter_fields()
.map(|(name, value)| (name.into(), value.clone()))
.collect();
fields.insert("resolution".into(), Value::text(choice.as_str()));
let mut publication = crate::storage::unified::store::RowPublication::default();
branches::stage_metadata(
&store,
&mut publication,
vc::CONFLICTS,
"id",
conflict_id,
fields,
)?;
store.publish_rows(publication)?;
Ok(())
}
pub fn vcs_resolve_as_of(&self, spec: AsOfSpec) -> RedDBResult<Xid> {
Ok(self.vcs_resolve_snapshot(spec)?.xid)
}
pub(super) fn vcs_resolve_snapshot(&self, spec: AsOfSpec) -> RedDBResult<Snapshot> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
match spec {
AsOfSpec::Snapshot(xid) => Ok(Snapshot {
xid,
in_progress: HashSet::new(),
}),
AsOfSpec::Commit(h) => {
let entity = load_commit_entity(store, &h)?
.ok_or_else(|| RedDBError::NotFound(format!("commit `{h}` not found")))?;
let row = entity.data.as_row().ok_or_else(|| {
RedDBError::Internal("vcs: invalid commit record".to_string())
})?;
commit_snapshot_from_row(row)
}
AsOfSpec::Branch(name) => {
let full = normalize_branch_name(&name);
let r = load_ref(store, &full)?.ok_or_else(|| {
RedDBError::NotFound(format!("branch `{full}` does not exist"))
})?;
let entity = load_commit_entity(store, &r.target)?.ok_or_else(|| {
RedDBError::NotFound(format!("branch `{full}` points to missing commit"))
})?;
let row = entity.data.as_row().ok_or_else(|| {
RedDBError::Internal("vcs: invalid commit record".to_string())
})?;
commit_snapshot_from_row(row)
}
AsOfSpec::Tag(name) => {
let full = normalize_tag_name(&name);
let r = load_ref(store, &full)?
.ok_or_else(|| RedDBError::NotFound(format!("tag `{full}` does not exist")))?;
let entity = load_commit_entity(store, &r.target)?.ok_or_else(|| {
RedDBError::NotFound(format!("tag `{full}` points to missing commit"))
})?;
let row = entity.data.as_row().ok_or_else(|| {
RedDBError::Internal("vcs: invalid commit record".to_string())
})?;
commit_snapshot_from_row(row)
}
AsOfSpec::TimestampMs(ts) => {
let manager = store
.get_collection(vc::COMMITS)
.ok_or_else(|| RedDBError::NotFound("no commits exist yet".to_string()))?;
let mut best: Option<(i64, UnifiedEntity)> = None;
let mut entities = Vec::new();
manager.try_for_each_entity(usize::MAX, |entity| {
entities.push(entity.clone());
true
})?;
for entity in entities {
if let Some(row) = entity.data.as_row() {
let t = row_i64(row, "timestamp_ms").unwrap_or(0);
if t <= ts {
match &best {
Some((bt, _)) if *bt >= t => {}
_ => best = Some((t, entity)),
}
}
}
}
let (_, entity) = best.ok_or_else(|| {
RedDBError::NotFound(format!("no commit at or before ts={ts}"))
})?;
let row = entity.data.as_row().ok_or_else(|| {
RedDBError::Internal("vcs: invalid commit record".to_string())
})?;
commit_snapshot_from_row(row)
}
}
}
pub fn vcs_set_versioned(&self, collection: &str, enabled: bool) -> RedDBResult<()> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let _publication = store.vault_publication_lock.lock();
if collection.starts_with("red_") || collection.starts_with("red.") {
return Err(RedDBError::InvalidConfig(format!(
"cannot version internal collection `{collection}`"
)));
}
if is_versioned(store, collection)? == enabled {
return Ok(());
}
if self.snapshot_manager().oldest_active_xid().is_some() {
return Err(RedDBError::InvalidOperation(
"finish active transactions before changing version history".into(),
));
}
if !enabled
&& self
.db()
.collection_contract(collection)
.is_some_and(|contract| {
matches!(
contract.declared_model,
crate::catalog::CollectionModel::Hll
| crate::catalog::CollectionModel::Sketch
| crate::catalog::CollectionModel::Filter
)
})
{
return Err(RedDBError::InvalidOperation(
"export probabilistic state before discarding its version history".into(),
));
}
if !enabled && list_refs_by_prefix(store, Some(vc::BRANCH_REF_PREFIX))?.len() > 1 {
return Err(RedDBError::InvalidOperation(
"cannot disable history while branches share versions".into(),
));
}
if enabled && store.format_version() < reddb_file::native_store::STORE_VERSION_V14 {
let previous_format = store.format_version();
store.set_format_version(reddb_file::native_store::STORE_VERSION_V14);
if store.pager().is_some() {
if let Err(error) = store.persist() {
store.set_format_version(previous_format);
return Err(RedDBError::Internal(error.to_string()));
}
}
}
let mut publication = crate::storage::unified::store::RowPublication::default();
if enabled {
if let Some(mut entity) = self.vcs_initial_probabilistic_state(collection)? {
entity.id = store.next_entity_id();
publication.inserts.push((collection.into(), entity, None));
}
}
if self
.db()
.collection_contract(collection)
.is_some_and(|contract| {
contract.declared_model == crate::catalog::CollectionModel::Queue
})
{
super::vcs_queue::stage_metadata_scope(store, &mut publication, collection, enabled)?;
}
if let Some(manager) = store.get_collection(collection) {
if !enabled {
let mut shared = false;
manager.try_for_each_entity(usize::MAX, |entity| {
shared = entity.branch_version().is_some_and(|version| {
version.origin != 0 || !version.deletions.is_empty()
});
!shared
})?;
if shared {
return Err(RedDBError::InvalidOperation("cannot discard branch provenance; export the selected branch to an unversioned collection".into()));
}
}
let mut ids = Vec::new();
manager.try_for_each_entity(usize::MAX, |entity| {
if entity.branch_version().is_some() != enabled {
ids.push(entity.id);
}
true
})?;
for id in ids {
let mut entity = manager
.try_get(id, usize::MAX)?
.ok_or_else(|| RedDBError::NotFound(id.to_string()))?;
if enabled {
entity.set_branch_origin(0);
} else {
entity.clear_branch_version();
}
let metadata = store.try_get_metadata(collection, id, usize::MAX)?;
publication
.replacements
.push((collection.into(), entity, metadata));
}
}
branches::stage_metadata(
store,
&mut publication,
vc::SETTINGS,
"id",
collection,
HashMap::from([
("id".into(), Value::text(collection)),
("versioned".into(), Value::Boolean(enabled)),
("ts_ms".into(), Value::TimestampMs(now_ms())),
]),
)?;
store.publish_rows(publication)?;
self.invalidate_result_cache();
Ok(())
}
pub fn vcs_list_versioned(&self) -> RedDBResult<Vec<String>> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
let mut list = versioned_collections(store)?;
list.sort();
Ok(list)
}
pub fn vcs_is_versioned(&self, collection: &str) -> RedDBResult<bool> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
is_versioned(store, collection)
}
pub fn vcs_resolve_commitish(&self, spec: &str) -> RedDBResult<CommitHash> {
let store_arc = self.inner.db.store();
let store: &UnifiedStore = &store_arc;
if spec.is_empty() {
return Err(RedDBError::InvalidConfig("empty commitish".to_string()));
}
if spec.len() == 64
&& spec.chars().all(|c| c.is_ascii_hexdigit())
&& load_commit(store, spec)?.is_some()
{
return Ok(spec.to_string());
}
if spec.starts_with("refs/") {
if let Some(hash) = resolve_ref_chain(store, spec)? {
return Ok(hash);
}
}
let normalized_branch = normalize_branch_name(spec);
if let Some(hash) = resolve_ref_chain(store, &normalized_branch)? {
return Ok(hash);
}
let normalized_tag = normalize_tag_name(spec);
if let Some(hash) = resolve_ref_chain(store, &normalized_tag)? {
return Ok(hash);
}
if let Some(hash) = resolve_short_commit(store, spec)? {
return Ok(hash);
}
Err(RedDBError::NotFound(format!(
"commitish `{spec}` did not resolve to any ref or commit"
)))
}
}
fn stage_merge_conflicts(
store: &UnifiedStore,
publication: &mut crate::storage::unified::store::RowPublication,
conflicts: &[Conflict],
) -> RedDBResult<()> {
for conflict in conflicts {
let fields = HashMap::from([
("id".into(), Value::text(conflict.id.as_str())),
(
"collection".into(),
Value::text(conflict.collection.as_str()),
),
("entity_id".into(), Value::text(conflict.entity_id.as_str())),
(
"merge_state_id".into(),
Value::text(conflict.merge_state_id.as_str()),
),
(
"conflicting_paths".into(),
Value::Array(
conflict
.conflicting_paths
.iter()
.map(|path| Value::text(path.as_str()))
.collect(),
),
),
(
"base_json".into(),
Value::Json(
crate::json::to_vec(&conflict.base)
.map_err(|e| RedDBError::Internal(e.to_string()))?,
),
),
(
"ours_json".into(),
Value::Json(
crate::json::to_vec(&conflict.ours)
.map_err(|e| RedDBError::Internal(e.to_string()))?,
),
),
(
"theirs_json".into(),
Value::Json(
crate::json::to_vec(&conflict.theirs)
.map_err(|e| RedDBError::Internal(e.to_string()))?,
),
),
]);
branches::stage_metadata(
store,
publication,
vc::CONFLICTS,
"id",
&conflict.id,
fields,
)?;
}
Ok(())
}
fn ensure_vcs_entity_access(
runtime: &RedDBRuntime,
collection: &str,
entity: &UnifiedEntity,
action: reddb_rql::ast::PolicyAction,
) -> RedDBResult<()> {
use reddb_rql::ast::PolicyAction;
runtime.check_collection_privilege(action, collection)?;
if !crate::runtime::rls_injection::entity_passes_rls_action(runtime, collection, action, entity)
{
return Err(RedDBError::Query(format!(
"VCS access to collection '{collection}' denied by row-level security"
)));
}
if let Some(row) = entity.data.as_row() {
if let Some(key) = row_text(row, "key") {
match runtime
.db()
.collection_contract(collection)
.map(|c| c.declared_model)
{
Some(crate::catalog::CollectionModel::Vault) => {
let capability = if action == PolicyAction::Select {
"vault:read_metadata"
} else {
"vault:write"
};
runtime
.check_vault_capability_in_tenant(
capability,
collection,
&key,
row_text(row, "_vault_tenant").as_deref(),
)
.map_err(RedDBError::Query)?;
}
Some(crate::catalog::CollectionModel::Config) => {
let capability = if action == PolicyAction::Select {
"config:read"
} else {
"config:write"
};
runtime
.check_config_capability(capability, collection, &key)
.map_err(RedDBError::Query)?;
}
_ => {}
}
}
}
Ok(())
}
fn vcs_entity_body(runtime: &RedDBRuntime, collection: &str, entity: &UnifiedEntity) -> JsonValue {
if runtime
.db()
.collection_contract(collection)
.is_some_and(|contract| contract.declared_model == crate::catalog::CollectionModel::Vault)
{
let mut fields = Map::new();
fields.insert(
"id".into(),
JsonValue::String(entity.logical_id().raw().to_string()),
);
if let Some(row) = entity.data.as_row() {
for name in ["key", "version", "tombstone"] {
if let Some(value) = row.get_field(name) {
fields.insert(name.into(), value.to_json());
}
}
}
return JsonValue::Object(fields);
}
let mut body = crate::entity_render::compact_entity_json(entity);
if let JsonValue::Object(fields) = &mut body {
fields.insert(
"id".to_string(),
JsonValue::Number(entity.logical_id().raw() as f64),
);
match &entity.kind {
EntityKind::GraphNode(node) => {
fields.insert("label".to_string(), JsonValue::String(node.label.clone()));
fields.insert(
"node_type".to_string(),
JsonValue::String(node.node_type.clone()),
);
}
EntityKind::GraphEdge(edge) => {
fields.insert("label".to_string(), JsonValue::String(edge.label.clone()));
fields.insert(
"from_node".to_string(),
JsonValue::String(edge.from_node.clone()),
);
fields.insert(
"to_node".to_string(),
JsonValue::String(edge.to_node.clone()),
);
}
_ => {}
}
if let EntityData::Row(row) = &entity.data {
if row.named.is_none() {
fields.insert(
"row".to_string(),
JsonValue::Object(
row.iter_fields()
.map(|(key, value)| (key.to_string(), value.to_json()))
.collect(),
),
);
}
}
}
body
}
fn coalesce_modifications(
entries: Vec<DiffEntry>,
added: &mut usize,
removed: &mut usize,
modified: &mut usize,
) -> Vec<DiffEntry> {
let mut by_key: HashMap<(String, String), Vec<DiffEntry>> = HashMap::new();
for e in entries {
by_key
.entry((e.collection.clone(), e.entity_id.clone()))
.or_default()
.push(e);
}
let mut out: Vec<DiffEntry> = Vec::new();
for ((coll, eid), group) in by_key {
if group.len() >= 2 {
let mut before = JsonValue::Null;
let mut after = JsonValue::Null;
for item in group {
match item.change {
DiffChange::Removed { before: b } => before = b,
DiffChange::Added { after: a } => after = a,
DiffChange::Modified { .. } => {}
}
}
*added = added.saturating_sub(1);
*removed = removed.saturating_sub(1);
*modified += 1;
out.push(DiffEntry {
collection: coll,
entity_id: eid,
change: DiffChange::Modified { before, after },
});
} else {
out.extend(group);
}
}
out.sort_by(|a, b| {
a.collection
.cmp(&b.collection)
.then(a.entity_id.cmp(&b.entity_id))
});
out
}
#[allow(dead_code)]
fn _touch_types(_: BTreeSet<()>) {}