use crate::table::{Table, TableError};
use super::entry::BufferEntry;
use super::merge::{apply_clean_entries, Conflict, MergeResult};
use super::resolve::{resolve, Resolution};
#[derive(Debug, Clone, PartialEq)]
pub struct MergeSession {
clean: Vec<BufferEntry>,
pending: Vec<Conflict>,
resolved: Vec<BufferEntry>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct MergeSessionResult {
pub applied: usize,
pub conflicts: Vec<Conflict>,
}
#[derive(Debug)]
pub enum MergeSessionError {
UnresolvedConflicts { remaining: usize },
ApplyFailed(TableError),
}
impl std::fmt::Display for MergeSessionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
MergeSessionError::UnresolvedConflicts { remaining } => {
write!(f, "{remaining} conflict(s) still unresolved")
}
MergeSessionError::ApplyFailed(cause) => cause.fmt(f),
}
}
}
impl std::error::Error for MergeSessionError {}
impl MergeSession {
pub fn new(result: MergeResult) -> Self {
MergeSession {
clean: result.clean,
pending: result.conflicts,
resolved: Vec::new(),
}
}
pub fn pending_conflicts(&self) -> &[Conflict] {
&self.pending
}
pub fn resolve_one(&mut self, index: usize, resolution: Resolution) {
let conflict = self.pending.remove(index);
let resolved_entry = resolve(&conflict, resolution);
self.resolved.push(resolved_entry);
}
pub fn is_fully_resolved(&self) -> bool {
self.pending.is_empty()
}
pub fn continue_merge(self, table: &mut Table) -> Result<MergeSessionResult, MergeSessionError> {
if !self.is_fully_resolved() {
return Err(MergeSessionError::UnresolvedConflicts {
remaining: self.pending.len(),
});
}
let mut entries = self.clean;
entries.extend(self.resolved);
let applied = entries.len();
let conflicts = apply_clean_entries(table, entries).map_err(|cause| {
MergeSessionError::ApplyFailed(cause)
})?;
Ok(MergeSessionResult { applied, conflicts })
}
}
fn render_entry(entry: &BufferEntry) -> String {
let op = match entry.op {
super::entry::BufferOp::Put => "put",
super::entry::BufferOp::Patch => "patch",
super::entry::BufferOp::Delete => "delete",
};
let body = serde_json::json!({ "op": op, "data": entry.data });
serde_json::to_string_pretty(&body).unwrap_or_default()
}
pub fn format_conflict_markers(conflict: &Conflict, base: Option<&BufferEntry>) -> String {
let mut lines = vec![format!("conflict on {}.{}", conflict.table, conflict.key)];
if let Some(base_entry) = base {
lines.push("||||||| base".to_string());
lines.push(render_entry(base_entry));
}
let last = conflict.versions.len().saturating_sub(1);
for (i, (source, entry)) in conflict.versions.iter().enumerate() {
if i == 0 {
lines.push(format!("<<<<<<< {source}"));
} else {
lines.push("=======".to_string());
}
lines.push(render_entry(entry));
if i == last {
lines.push(format!(">>>>>>> {source}"));
}
}
lines.join("\n")
}
#[cfg(test)]
mod tests {
use serde_json::{json, Map};
use tempfile::tempdir;
use super::*;
use crate::buffer::entry::BufferOp;
use crate::schema::parse_schema;
fn users_schema() -> crate::schema::TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
},
}))
.unwrap()
}
fn entry(key: &str, source: &str, buffered_at: i64, name: &str) -> BufferEntry {
let mut data = Map::new();
data.insert("id".into(), json!(key));
data.insert("name".into(), json!(name));
BufferEntry {
table: "users".into(),
key: key.into(),
op: BufferOp::Put,
data: Some(data),
buffered_at,
source: source.into(),
}
}
fn conflict(key: &str) -> Conflict {
Conflict {
table: "users".into(),
key: key.into(),
versions: vec![
("branch-a".into(), entry(key, "branch-a", 100, "Ania")),
("branch-b".into(), entry(key, "branch-b", 200, "Zosia")),
],
}
}
fn session_with_two_conflicts() -> MergeSession {
let clean = vec![entry("u3", "branch-a", 50, "Kasia")];
let conflicts = vec![conflict("u1"), conflict("u2")];
MergeSession::new(MergeResult { clean, conflicts })
}
#[test]
fn resolving_one_of_two_conflicts_leaves_the_session_unresolved() {
let mut session = session_with_two_conflicts();
assert_eq!(session.pending_conflicts().len(), 2);
session.resolve_one(0, Resolution::Ours("branch-a".into()));
assert!(!session.is_fully_resolved());
assert_eq!(session.pending_conflicts().len(), 1);
session.resolve_one(0, Resolution::Theirs("branch-b".into()));
assert!(session.is_fully_resolved());
}
#[test]
fn continue_merge_refuses_and_writes_nothing_before_full_resolution() {
let dir = tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut table = Table::open(users_schema(), &path).unwrap();
let seq_before = table.current_seq();
let mut session = session_with_two_conflicts();
session.resolve_one(0, Resolution::Ours("branch-a".into()));
let result = session.continue_merge(&mut table);
assert!(matches!(
result,
Err(MergeSessionError::UnresolvedConflicts { remaining: 1 })
));
assert_eq!(table.current_seq(), seq_before);
assert_eq!(std::fs::read_to_string(&path).unwrap(), "");
}
#[test]
fn continue_merge_applies_clean_and_resolved_entries_once_fully_resolved() {
let dir = tempdir().unwrap();
let path = dir.path().join("users.jsonl");
let mut table = Table::open(users_schema(), &path).unwrap();
let mut session = session_with_two_conflicts();
session.resolve_one(0, Resolution::Ours("branch-a".into()));
session.resolve_one(0, Resolution::Theirs("branch-b".into()));
assert!(session.is_fully_resolved());
let result = session.continue_merge(&mut table).unwrap();
assert_eq!(result.applied, 3);
assert!(result.conflicts.is_empty());
assert!(table.get("u1").unwrap().is_some());
assert!(table.get("u2").unwrap().is_some());
assert!(table.get("u3").unwrap().is_some());
}
#[test]
fn format_conflict_markers_shows_every_version_between_markers() {
let rendered = format_conflict_markers(&conflict("u1"), None);
assert!(rendered.contains("<<<<<<< branch-a"));
assert!(rendered.contains("======="));
assert!(rendered.contains(">>>>>>> branch-b"));
assert!(rendered.contains("Ania"));
assert!(rendered.contains("Zosia"));
assert!(!rendered.contains("|||||||"));
}
#[test]
fn format_conflict_markers_adds_a_base_section_when_a_base_is_given() {
let base = entry("u1", "base", 0, "Original");
let rendered = format_conflict_markers(&conflict("u1"), Some(&base));
assert!(rendered.contains("||||||| base"));
assert!(rendered.contains("Original"));
}
}