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