grove
grove/src/store.rs
at @4
1//! The store: objects, the log, and the state the log adds up to.
2//!
3//! A write is: check the referenced objects exist, apply the record to the in-memory state
4//! (which validates it and changes nothing if it's invalid), then append it to the log. If
5//! the append fails, the state is rebuilt from the log on disk, so memory never claims
6//! something the disk doesn't hold.
7
8use crate::json::{self, Value};
9use crate::log::{Log, OpenError, Recovery};
10use crate::model::{id_to_json, step_to_json, Actor, ChangeState, Project, State, Step};
11use crate::objects::{self, Id, Objects};
12use std::path::{Path, PathBuf};
13use std::time::{SystemTime, UNIX_EPOCH};
14
15#[derive(Debug)]
16pub enum Error {
17 Open(OpenError),
18 Objects(objects::Error),
19 Io(std::io::Error),
20 /// The log holds a record that doesn't fit the state before it.
21 Damaged { record: u64, reason: String },
22 /// The request isn't allowed or doesn't make sense.
23 Refused(String),
24 /// The change can't be rebuilt on the current tip without someone resolving these paths.
25 Conflict { step: usize, paths: Vec<String> },
26}
27
28impl std::fmt::Display for Error {
29 fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
30 match self {
31 Error::Open(e) => write!(f, "{e}"),
32 Error::Objects(e) => write!(f, "{e}"),
33 Error::Io(e) => write!(f, "{e}"),
34 Error::Damaged { record, reason } => write!(f, "log record {record} is damaged: {reason}"),
35 Error::Refused(r) => write!(f, "{r}"),
36 Error::Conflict { step, paths } => write!(f, "step {} conflicts with main in {}", step + 1, paths.join(", ")),
37 }
38 }
39}
40
41impl From<OpenError> for Error {
42 fn from(e: OpenError) -> Self {
43 Error::Open(e)
44 }
45}
46impl From<objects::Error> for Error {
47 fn from(e: objects::Error) -> Self {
48 Error::Objects(e)
49 }
50}
51impl From<std::io::Error> for Error {
52 fn from(e: std::io::Error) -> Self {
53 Error::Io(e)
54 }
55}
56
57pub type Result<T> = std::result::Result<T, Error>;
58
59fn refused<T>(msg: impl Into<String>) -> Result<T> {
60 Err(Error::Refused(msg.into()))
61}
62
63/// What a piece of evidence holds.
64pub enum Content<'a> {
65 Bytes(&'a [u8]),
66 Url(&'a str),
67}
68
69pub fn now_ms() -> i64 {
70 SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)
71}
72
73pub struct Store {
74 dir: PathBuf,
75 log: Log,
76 pub objects: Objects,
77 state: State,
78 pub recovery: Recovery,
79}
80
81fn replay(records: &[Vec<u8>]) -> Result<State> {
82 let mut state = State::default();
83 for (i, bytes) in records.iter().enumerate() {
84 let damaged = |reason: String| Error::Damaged { record: i as u64, reason };
85 let text = std::str::from_utf8(bytes).map_err(|_| damaged("not UTF-8".into()))?;
86 let rec = json::parse(text).map_err(|e| damaged(e.to_string()))?;
87 state.apply(&rec).map_err(damaged)?;
88 }
89 Ok(state)
90}
91
92impl Store {
93 pub fn open(dir: &Path) -> Result<Store> {
94 std::fs::create_dir_all(dir)?;
95 let objects = Objects::open(&dir.join("objects"))?;
96 let (log, records, recovery) = Log::open(&dir.join("log"))?;
97 let state = replay(&records)?;
98 Ok(Store { dir: dir.to_path_buf(), log, objects, state, recovery })
99 }
100
101 pub fn dir(&self) -> &Path {
102 &self.dir
103 }
104
105 pub fn state(&self) -> &State {
106 &self.state
107 }
108
109 /// Validate, apply, append. `rec` gets its time and actor here.
110 fn write(&mut self, by: &Actor, mut rec: Vec<(String, Value)>) -> Result<()> {
111 rec.insert(1, ("at".into(), now_ms().into()));
112 rec.insert(2, ("by".into(), by.to_json()));
113 let rec = Value::Object(rec);
114 self.state.apply(&rec).map_err(Error::Refused)?;
115 if let Err(e) = self.log.append(rec.encode().as_bytes()) {
116 // The state now holds a record the log doesn't. Rebuild it from the disk.
117 let (log, records, _) = Log::open(&self.dir.join("log")).map_err(|_| Error::Io(e))?;
118 self.log = log;
119 self.state = replay(&records)?;
120 return Err(Error::Refused("couldn't write to the log".into()));
121 }
122 Ok(())
123 }
124
125 fn rec(t: &str, fields: Vec<(&str, Value)>) -> Vec<(String, Value)> {
126 let mut out = vec![("t".to_string(), Value::from(t))];
127 out.extend(fields.into_iter().map(|(k, v)| (k.to_string(), v)));
128 out
129 }
130
131 fn require_tree(&self, id: &Id) -> Result<()> {
132 if !self.objects.has(id) {
133 return refused(format!("snapshot {} isn't stored", crate::hex::encode(id)));
134 }
135 Ok(())
136 }
137
138 pub fn project(&self, owner: &str, name: &str) -> Result<&Project> {
139 self.state.project(owner, name).ok_or_else(|| Error::Refused(format!("no project {owner}/{name}")))
140 }
141
142 // ── People and agents ──────────────────────────────────
143
144 pub fn add_person(&mut self, username: &str, display_name: &str) -> Result<()> {
145 if !username.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') || username.is_empty() {
146 return refused("usernames are letters, digits, - and _");
147 }
148 self.write(&Actor::person(username), Self::rec("person", vec![("username", username.into()), ("display_name", display_name.into())]))
149 }
150
151 /// Register an agent acting for `person`. Returns its id.
152 pub fn add_agent(&mut self, person: &str, name: &str) -> Result<String> {
153 let id = crate::random::hex(6);
154 self.write(&Actor::person(person), Self::rec("agent", vec![("id", id.clone().into()), ("name", name.into())]))?;
155 Ok(id)
156 }
157
158 pub fn revoke_agent(&mut self, person: &str, id: &str) -> Result<()> {
159 self.write(&Actor::person(person), Self::rec("agent.revoke", vec![("id", id.into())]))
160 }
161
162 // ── Projects and main ──────────────────────────────────
163
164 /// Create a project owned by `owner`: the actor themselves or a group they're in.
165 pub fn create_project(&mut self, by: &Actor, owner: &str, name: &str, public: bool) -> Result<()> {
166 if !objects::valid_name(name) || name.contains(char::is_whitespace) {
167 return refused(format!("{name:?} can't be a project name"));
168 }
169 self.write(by, Self::rec("project", vec![("owner", owner.into()), ("name", name.into()), ("public", public.into())]))
170 }
171
172 /// Give a project to another owner. History goes with it.
173 pub fn move_project(&mut self, by: &Actor, owner: &str, name: &str, to: &str) -> Result<()> {
174 self.write(by, Self::rec("project.move", vec![("owner", owner.into()), ("project", name.into()), ("to", to.into())]))
175 }
176
177 pub fn create_group(&mut self, by: &Actor, name: &str) -> Result<()> {
178 if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') || name.is_empty() {
179 return refused("group names are letters, digits, - and _");
180 }
181 self.write(by, Self::rec("group", vec![("name", name.into())]))
182 }
183
184 pub fn add_member(&mut self, by: &Actor, group: &str, person: &str) -> Result<()> {
185 self.write(by, Self::rec("group.add", vec![("group", group.into()), ("person", person.into())]))
186 }
187
188 /// Record a point directly on main: a project's first snapshot, or imported history.
189 pub fn record_point(&mut self, by: &Actor, owner: &str, name: &str, snapshot: &Id, message: &str, author: Option<&Actor>) -> Result<u64> {
190 self.require_tree(snapshot)?;
191 let n = self.project(owner, name)?.tip() + 1;
192 let mut fields = vec![
193 ("owner", owner.into()),
194 ("project", name.into()),
195 ("n", n.into()),
196 ("snapshot", id_to_json(snapshot)),
197 ("message", message.into()),
198 ];
199 if let Some(a) = author {
200 fields.push(("author", a.to_json()));
201 }
202 self.write(by, Self::rec("point", fields))?;
203 Ok(n)
204 }
205
206 // ── Changes ────────────────────────────────────────────
207
208 /// Open a change as an idea: an intent, no code yet. Its first revision makes it a draft.
209 pub fn open_change(&mut self, by: &Actor, owner: &str, name: &str, intent: &str) -> Result<u64> {
210 let p = self.project(owner, name)?;
211 if !p.public && !self.state.owns(&p.owner, &by.person) {
212 return refused(format!("{owner}/{name} is private"));
213 }
214 let number = p.changes.len() as u64 + 1;
215 self.write(by, Self::rec("change", vec![("owner", owner.into()), ("project", name.into()), ("number", number.into()), ("intent", intent.into()), ("state", "idea".into())]))?;
216 Ok(number)
217 }
218
219 /// Add a revision: the whole stack of steps, built on main point `base`.
220 pub fn revise(&mut self, by: &Actor, owner: &str, name: &str, change: u64, base: u64, steps: &[Step], note: Option<&str>) -> Result<u32> {
221 for s in steps {
222 self.require_tree(&s.snapshot)?;
223 }
224 let p = self.project(owner, name)?;
225 let c = p.change(change).ok_or_else(|| Error::Refused(format!("no change #{change}")))?;
226 if by.person != c.proposer.person && !self.state.owns(&p.owner, &by.person) {
227 return refused("only the proposer or the owner can revise a change");
228 }
229 let number = c.revisions.len() as u32 + 1;
230 let steps: Vec<Value> = steps.iter().map(step_to_json).collect();
231 self.write(
232 by,
233 Self::rec("revision", vec![
234 ("owner", owner.into()),
235 ("project", name.into()),
236 ("change", change.into()),
237 ("number", (number as u64).into()),
238 ("base", base.into()),
239 ("steps", steps.into()),
240 ("note", note.map(str::to_string).into()),
241 ]),
242 )?;
243 Ok(number)
244 }
245
246 pub fn set_state(&mut self, by: &Actor, owner: &str, name: &str, change: u64, to: ChangeState) -> Result<()> {
247 if to == ChangeState::Landed {
248 return refused("changes land through land()");
249 }
250 self.write(by, Self::rec("state", vec![("owner", owner.into()), ("project", name.into()), ("change", change.into()), ("state", to.name().into())]))
251 }
252
253 /// Land a change's latest revision. Its steps become the next points on main, all at once.
254 pub fn land(&mut self, by: &Actor, owner: &str, name: &str, change: u64) -> Result<(u64, u64)> {
255 let p = self.project(owner, name)?;
256 if !self.state.owns(&p.owner, &by.person) || by.agent.is_some() {
257 return refused("only the owner, in person, lands changes");
258 }
259 let c = p.change(change).ok_or_else(|| Error::Refused(format!("no change #{change}")))?;
260 let rev = c.latest().ok_or_else(|| Error::Refused(format!("#{change} has no revisions")))?;
261 let tip = p.tip();
262 let steps: Vec<Id> = rev.steps.iter().map(|s| s.snapshot).collect();
263 // Built on an older point: replay each step's edits onto the current tip.
264 let snapshots = if rev.base == tip {
265 steps
266 } else {
267 let from = p.point(rev.base).map(|pt| pt.snapshot).ok_or_else(|| Error::Refused("base point missing".into()))?;
268 let onto = p.point(tip).map(|pt| pt.snapshot).unwrap();
269 match crate::merge::rebuild(&self.objects, &from, &steps, &onto)? {
270 Ok(rebuilt) => rebuilt,
271 Err((step, paths)) => return Err(Error::Conflict { step, paths }),
272 }
273 };
274 let first = tip + 1;
275 let count = snapshots.len() as u64;
276 let snapshots: Vec<Value> = snapshots.iter().map(id_to_json).collect();
277 let revision = rev.number as u64;
278 self.write(
279 by,
280 Self::rec("land", vec![
281 ("owner", owner.into()),
282 ("project", name.into()),
283 ("change", change.into()),
284 ("revision", revision.into()),
285 ("first", first.into()),
286 ("snapshots", snapshots.into()),
287 ]),
288 )?;
289 Ok((first, first + count - 1))
290 }
291
292 pub fn say(&mut self, by: &Actor, owner: &str, name: &str, change: u64, body: &str) -> Result<()> {
293 self.write(by, Self::rec("say", vec![("owner", owner.into()), ("project", name.into()), ("change", change.into()), ("body", body.into())]))
294 }
295
296 /// Attach evidence to a revision: stored bytes of a media type, or a URL.
297 pub fn attach(&mut self, by: &Actor, owner: &str, name: &str, change: u64, revision: u32, title: &str, media_type: &str, content: Content) -> Result<()> {
298 let (blob, url) = match content {
299 Content::Bytes(bytes) => (Value::from(crate::hex::encode(&self.objects.put_blob(bytes)?)), Value::Null),
300 Content::Url(url) => (Value::Null, Value::from(url)),
301 };
302 self.write(
303 by,
304 Self::rec("evidence", vec![
305 ("owner", owner.into()),
306 ("project", name.into()),
307 ("change", change.into()),
308 ("revision", (revision as u64).into()),
309 ("title", title.into()),
310 ("type", media_type.into()),
311 ("blob", blob),
312 ("url", url),
313 ]),
314 )
315 }
316
317 /// Check that every object the state refers to exists and matches its id.
318 pub fn verify(&self) -> Result<usize> {
319 let mut count = 0;
320 let mut seen = std::collections::HashSet::new();
321 let mut check_tree = |id: &Id, count: &mut usize| -> Result<()> {
322 if seen.insert(*id) {
323 *count += self.objects.verify_tree(id)?;
324 }
325 Ok(())
326 };
327 for p in &self.state.projects {
328 for pt in &p.points {
329 check_tree(&pt.snapshot, &mut count)?;
330 }
331 for c in &p.changes {
332 for r in &c.revisions {
333 for s in &r.steps {
334 check_tree(&s.snapshot, &mut count)?;
335 }
336 }
337 for e in &c.evidence {
338 if let Some(b) = &e.blob {
339 self.objects.get_blob(b)?;
340 count += 1;
341 }
342 }
343 }
344 }
345 Ok(count)
346 }
347}
348
349#[cfg(test)]
350mod tests {
351 use super::*;
352 use crate::objects::{Entry, Kind};
353 use crate::testdir::TestDir;
354
355 fn tree(store: &Store, files: &[(&str, &str)]) -> Id {
356 let entries: Vec<Entry> = files
357 .iter()
358 .map(|(n, c)| Entry { name: n.to_string(), kind: Kind::File, id: store.objects.put_blob(c.as_bytes()).unwrap() })
359 .collect();
360 store.objects.put_tree(&entries).unwrap()
361 }
362
363 fn setup(dir: &TestDir) -> (Store, Actor, String) {
364 let mut store = Store::open(dir.path()).unwrap();
365 store.add_person("anton", "Anton").unwrap();
366 let agent = store.add_agent("anton", "claude").unwrap();
367 let me = Actor::agent("anton", &agent);
368 store.create_project(&Actor::person("anton"), "anton", "grove", true).unwrap();
369 let t = tree(&store, &[("README", "hello")]);
370 store.record_point(&Actor::person("anton"), "anton", "grove", &t, "Start", None).unwrap();
371 (store, me, agent)
372 }
373
374 #[test]
375 fn change_lifecycle_and_replay() {
376 let dir = TestDir::new("store-lifecycle");
377 {
378 let (mut store, me, _) = setup(&dir);
379 let owner = Actor::person("anton");
380 let c = store.open_change(&me, "anton", "grove", "Say more in the README").unwrap();
381 assert_eq!(store.project("anton", "grove").unwrap().change(c).unwrap().state, ChangeState::Idea);
382
383 let t1 = tree(&store, &[("README", "hello there")]);
384 let t2 = tree(&store, &[("README", "hello there"), ("NOTES", "x")]);
385 let steps = vec![
386 Step { id: "s1".into(), snapshot: t1, message: "Longer greeting".into() },
387 Step { id: "s2".into(), snapshot: t2, message: "Add notes".into() },
388 ];
389 assert_eq!(store.revise(&me, "anton", "grove", c, 1, &steps, None).unwrap(), 1);
390 store.set_state(&me, "anton", "grove", c, ChangeState::Proposed).unwrap();
391 store.say(&owner, "anton", "grove", c, "Looks good").unwrap();
392 store.attach(&me, "anton", "grove", c, 1, "notes", "text/plain", Content::Bytes(b"checked by hand")).unwrap();
393
394 // Only the owner lands.
395 assert!(matches!(store.land(&me, "anton", "grove", c), Err(Error::Refused(_))));
396 assert_eq!(store.land(&owner, "anton", "grove", c).unwrap(), (2, 3));
397 let p = store.project("anton", "grove").unwrap();
398 assert_eq!(p.tip(), 3);
399 assert_eq!(p.point(3).unwrap().snapshot, t2);
400 assert_eq!(p.change(c).unwrap().state, ChangeState::Landed);
401 assert_eq!(store.verify().unwrap() > 0, true);
402 }
403 // Everything comes back from the log.
404 let store = Store::open(dir.path()).unwrap();
405 let p = store.project("anton", "grove").unwrap();
406 assert_eq!(p.tip(), 3);
407 let c = p.change(1).unwrap();
408 assert_eq!(c.state, ChangeState::Landed);
409 assert_eq!(c.messages[0].body, "Looks good");
410 assert_eq!(c.evidence.len(), 1);
411 assert_eq!(p.point(2).unwrap().message, "Longer greeting");
412 assert!(p.point(2).unwrap().author.agent.is_some());
413 }
414
415 #[test]
416 fn landing_on_a_moved_main() {
417 let dir = TestDir::new("store-behind");
418 let (mut store, me, _) = setup(&dir);
419 let owner = Actor::person("anton");
420 // Both changes start from @1 (README = "hello").
421 let ta = tree(&store, &[("README", "hello"), ("a.txt", "from a\n")]);
422 let tb = tree(&store, &[("README", "hello"), ("b.txt", "from b\n")]);
423 let tc = tree(&store, &[("README", "clash")]);
424 let a = store.open_change(&me, "anton", "grove", "a").unwrap();
425 let b = store.open_change(&me, "anton", "grove", "b").unwrap();
426 let c = store.open_change(&me, "anton", "grove", "c").unwrap();
427 let tl = tree(&store, &[("README", "landed")]);
428 for (n, t) in [(a, tl), (b, tb), (c, tc)] {
429 store.revise(&me, "anton", "grove", n, 1, &[Step { id: "s".into(), snapshot: t, message: "m".into() }], None).unwrap();
430 store.set_state(&me, "anton", "grove", n, ChangeState::Proposed).unwrap();
431 }
432 store.land(&owner, "anton", "grove", a).unwrap();
433 // b touched other files: it's rebuilt on @2 and lands.
434 assert_eq!(store.land(&owner, "anton", "grove", b).unwrap(), (3, 3));
435 let p = store.project("anton", "grove").unwrap();
436 assert_eq!(p.point(3).unwrap().snapshot, tree(&store, &[("README", "landed"), ("b.txt", "from b\n")]));
437 // c changed README differently from what landed: a conflict, and nothing changes.
438 let before = store.state().records;
439 assert!(matches!(store.land(&owner, "anton", "grove", c), Err(Error::Conflict { step: 0, .. })));
440 assert_eq!(store.state().records, before);
441 assert_eq!(store.project("anton", "grove").unwrap().change(c).unwrap().state, ChangeState::Proposed);
442 }
443
444 #[test]
445 fn refusals_leave_no_trace() {
446 let dir = TestDir::new("store-refusals");
447 let (mut store, me, agent) = setup(&dir);
448 let before = store.state().records;
449 // Unstored snapshot, unknown project, bad transition, revoked agent.
450 let ghost = crate::objects::blob_id(b"not stored");
451 assert!(store.record_point(&Actor::person("anton"), "anton", "grove", &ghost, "x", None).is_err());
452 assert!(store.open_change(&me, "anton", "nope", "x").is_err());
453 let c = store.open_change(&me, "anton", "grove", "idea").unwrap();
454 assert!(store.set_state(&me, "anton", "grove", c, ChangeState::Proposed).is_err(), "an idea can't be proposed");
455 store.revoke_agent("anton", &agent).unwrap();
456 assert!(store.say(&me, "anton", "grove", c, "still here?").is_err());
457 assert_eq!(store.state().records, before + 2, "only the change and the revocation were written");
458 drop(store);
459 assert_eq!(Store::open(dir.path()).unwrap().state().records, before + 2);
460 }
461
462 #[test]
463 fn groups_own_projects() {
464 let dir = TestDir::new("store-groups");
465 let (mut store, me, _) = setup(&dir);
466 let anton = Actor::person("anton");
467 store.add_person("bea", "Bea").unwrap();
468 let bea = Actor::person("bea");
469 assert!(store.create_group(&me, "letterpress").is_err(), "groups are created in person");
470 store.create_group(&anton, "letterpress").unwrap();
471 assert!(store.create_group(&anton, "bea").is_err(), "names are shared with people");
472 // Move grove to the group; its history goes with it.
473 assert!(store.move_project(&bea, "anton", "grove", "letterpress").is_err());
474 store.move_project(&anton, "anton", "grove", "letterpress").unwrap();
475 assert!(store.project("anton", "grove").is_err());
476 assert_eq!(store.project("letterpress", "grove").unwrap().tip(), 1);
477 // Members decide for the group; others don't.
478 let t = tree(&store, &[("README", "group")]);
479 let c = store.open_change(&bea, "letterpress", "grove", "from bea").unwrap();
480 store.revise(&bea, "letterpress", "grove", c, 1, &[Step { id: "s".into(), snapshot: t, message: "m".into() }], None).unwrap();
481 store.set_state(&bea, "letterpress", "grove", c, ChangeState::Proposed).unwrap();
482 assert!(store.land(&bea, "letterpress", "grove", c).is_err());
483 store.add_member(&anton, "letterpress", "bea").unwrap();
484 store.land(&bea, "letterpress", "grove", c).unwrap();
485 drop(store);
486 let store = Store::open(dir.path()).unwrap();
487 assert_eq!(store.project("letterpress", "grove").unwrap().tip(), 2);
488 assert_eq!(store.state().group("letterpress").unwrap().members, ["anton", "bea"]);
489 }
490
491 #[test]
492 fn strangers_and_private_projects() {
493 let dir = TestDir::new("store-strangers");
494 let (mut store, _, _) = setup(&dir);
495 store.add_person("visitor", "Visitor").unwrap();
496 let visitor = Actor::person("visitor");
497 // Anyone can propose to a public project, but not land.
498 let c = store.open_change(&visitor, "anton", "grove", "An idea from outside").unwrap();
499 let t = tree(&store, &[("README", "outside")]);
500 store.revise(&visitor, "anton", "grove", c, 1, &[Step { id: "s".into(), snapshot: t, message: "m".into() }], None).unwrap();
501 store.set_state(&visitor, "anton", "grove", c, ChangeState::Proposed).unwrap();
502 assert!(store.land(&visitor, "anton", "grove", c).is_err());
503 store.set_state(&Actor::person("anton"), "anton", "grove", c, ChangeState::Declined).unwrap();
504 // Private projects take changes only from the owner.
505 store.create_project(&Actor::person("anton"), "anton", "secret", false).unwrap();
506 assert!(store.open_change(&visitor, "anton", "secret", "let me in").is_err());
507 }
508}