1use crate::table::{Table, TableError};
2
3use super::entry::BufferEntry;
4use super::merge::{apply_clean_entries, Conflict, MergeResult};
5use super::resolve::{resolve, Resolution};
6
7#[derive(Debug, Clone, PartialEq)]
8pub struct MergeSession {
9 clean: Vec<BufferEntry>,
10 pending: Vec<Conflict>,
11 resolved: Vec<BufferEntry>,
12}
13
14#[derive(Debug, Clone, PartialEq)]
15pub struct MergeSessionResult {
16 pub applied: usize,
17 pub conflicts: Vec<Conflict>,
18}
19
20#[derive(Debug)]
21pub enum MergeSessionError {
22 UnresolvedConflicts { remaining: usize },
23 ApplyFailed(TableError),
24}
25
26impl std::fmt::Display for MergeSessionError {
27 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28 match self {
29 MergeSessionError::UnresolvedConflicts { remaining } => {
30 write!(f, "{remaining} conflict(s) still unresolved")
31 }
32 MergeSessionError::ApplyFailed(cause) => cause.fmt(f),
33 }
34 }
35}
36
37impl std::error::Error for MergeSessionError {}
38
39impl MergeSession {
40 pub fn new(result: MergeResult) -> Self {
41 MergeSession {
42 clean: result.clean,
43 pending: result.conflicts,
44 resolved: Vec::new(),
45 }
46 }
47
48 pub fn pending_conflicts(&self) -> &[Conflict] {
49 &self.pending
50 }
51
52 pub fn resolve_one(&mut self, index: usize, resolution: Resolution) {
53 let conflict = self.pending.remove(index);
54 let resolved_entry = resolve(&conflict, resolution);
55 self.resolved.push(resolved_entry);
56 }
57
58 pub fn is_fully_resolved(&self) -> bool {
59 self.pending.is_empty()
60 }
61
62 pub fn continue_merge(self, table: &mut Table) -> Result<MergeSessionResult, MergeSessionError> {
63 if !self.is_fully_resolved() {
64 return Err(MergeSessionError::UnresolvedConflicts {
65 remaining: self.pending.len(),
66 });
67 }
68
69 let mut entries = self.clean;
70 entries.extend(self.resolved);
71 let applied = entries.len();
72 let conflicts = apply_clean_entries(table, entries).map_err(|cause| {
73 MergeSessionError::ApplyFailed(cause)
75 })?;
76 Ok(MergeSessionResult { applied, conflicts })
77 }
78}
79
80fn render_entry(entry: &BufferEntry) -> String {
81 let op = match entry.op {
82 super::entry::BufferOp::Put => "put",
83 super::entry::BufferOp::Patch => "patch",
84 super::entry::BufferOp::Delete => "delete",
85 };
86 let body = serde_json::json!({ "op": op, "data": entry.data });
87 serde_json::to_string_pretty(&body).unwrap_or_default()
88}
89
90pub fn format_conflict_markers(conflict: &Conflict, base: Option<&BufferEntry>) -> String {
91 let mut lines = vec![format!("conflict on {}.{}", conflict.table, conflict.key)];
92
93 if let Some(base_entry) = base {
94 lines.push("||||||| base".to_string());
95 lines.push(render_entry(base_entry));
96 }
97
98 let last = conflict.versions.len().saturating_sub(1);
99 for (i, (source, entry)) in conflict.versions.iter().enumerate() {
100 if i == 0 {
101 lines.push(format!("<<<<<<< {source}"));
102 } else {
103 lines.push("=======".to_string());
104 }
105 lines.push(render_entry(entry));
106 if i == last {
107 lines.push(format!(">>>>>>> {source}"));
108 }
109 }
110
111 lines.join("\n")
112}
113
114#[cfg(test)]
115mod tests {
116 use serde_json::{json, Map};
117 use tempfile::tempdir;
118
119 use super::*;
120 use crate::buffer::entry::BufferOp;
121 use crate::schema::parse_schema;
122
123 fn users_schema() -> crate::schema::TableSchema {
124 parse_schema(&json!({
125 "schemaVersion": 1,
126 "table": "users",
127 "primaryKey": "id",
128 "fields": {
129 "id": { "type": "string" },
130 "name": { "type": "string", "required": true },
131 },
132 }))
133 .unwrap()
134 }
135
136 fn entry(key: &str, source: &str, buffered_at: i64, name: &str) -> BufferEntry {
137 let mut data = Map::new();
138 data.insert("id".into(), json!(key));
139 data.insert("name".into(), json!(name));
140 BufferEntry {
141 table: "users".into(),
142 key: key.into(),
143 op: BufferOp::Put,
144 data: Some(data),
145 buffered_at,
146 source: source.into(),
147 }
148 }
149
150 fn conflict(key: &str) -> Conflict {
151 Conflict {
152 table: "users".into(),
153 key: key.into(),
154 versions: vec![
155 ("branch-a".into(), entry(key, "branch-a", 100, "Ania")),
156 ("branch-b".into(), entry(key, "branch-b", 200, "Zosia")),
157 ],
158 }
159 }
160
161 fn session_with_two_conflicts() -> MergeSession {
162 let clean = vec![entry("u3", "branch-a", 50, "Kasia")];
163 let conflicts = vec![conflict("u1"), conflict("u2")];
164 MergeSession::new(MergeResult { clean, conflicts })
165 }
166
167 #[test]
168 fn resolving_one_of_two_conflicts_leaves_the_session_unresolved() {
169 let mut session = session_with_two_conflicts();
170 assert_eq!(session.pending_conflicts().len(), 2);
171
172 session.resolve_one(0, Resolution::Ours("branch-a".into()));
173 assert!(!session.is_fully_resolved());
174 assert_eq!(session.pending_conflicts().len(), 1);
175
176 session.resolve_one(0, Resolution::Theirs("branch-b".into()));
177 assert!(session.is_fully_resolved());
178 }
179
180 #[test]
181 fn continue_merge_refuses_and_writes_nothing_before_full_resolution() {
182 let dir = tempdir().unwrap();
183 let path = dir.path().join("users.jsonl");
184 let mut table = Table::open(users_schema(), &path).unwrap();
185 let seq_before = table.current_seq();
186
187 let mut session = session_with_two_conflicts();
188 session.resolve_one(0, Resolution::Ours("branch-a".into()));
189
190 let result = session.continue_merge(&mut table);
191 assert!(matches!(
192 result,
193 Err(MergeSessionError::UnresolvedConflicts { remaining: 1 })
194 ));
195 assert_eq!(table.current_seq(), seq_before);
196 assert_eq!(std::fs::read_to_string(&path).unwrap(), "");
197 }
198
199 #[test]
200 fn continue_merge_applies_clean_and_resolved_entries_once_fully_resolved() {
201 let dir = tempdir().unwrap();
202 let path = dir.path().join("users.jsonl");
203 let mut table = Table::open(users_schema(), &path).unwrap();
204
205 let mut session = session_with_two_conflicts();
206 session.resolve_one(0, Resolution::Ours("branch-a".into()));
207 session.resolve_one(0, Resolution::Theirs("branch-b".into()));
208 assert!(session.is_fully_resolved());
209
210 let result = session.continue_merge(&mut table).unwrap();
211 assert_eq!(result.applied, 3);
212 assert!(result.conflicts.is_empty());
213 assert!(table.get("u1").unwrap().is_some());
214 assert!(table.get("u2").unwrap().is_some());
215 assert!(table.get("u3").unwrap().is_some());
216 }
217
218 #[test]
219 fn format_conflict_markers_shows_every_version_between_markers() {
220 let rendered = format_conflict_markers(&conflict("u1"), None);
221 assert!(rendered.contains("<<<<<<< branch-a"));
222 assert!(rendered.contains("======="));
223 assert!(rendered.contains(">>>>>>> branch-b"));
224 assert!(rendered.contains("Ania"));
225 assert!(rendered.contains("Zosia"));
226 assert!(!rendered.contains("|||||||"));
227 }
228
229 #[test]
230 fn format_conflict_markers_adds_a_base_section_when_a_base_is_given() {
231 let base = entry("u1", "base", 0, "Original");
232 let rendered = format_conflict_markers(&conflict("u1"), Some(&base));
233 assert!(rendered.contains("||||||| base"));
234 assert!(rendered.contains("Original"));
235 }
236}