grove
grove/src/store.rs
at @2
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 /// Landing needs the change rebuilt on the current tip first.
25 Behind { base: u64, tip: u64 },
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::Behind { base, tip } => write!(f, "built on point {base}, but main is at {tip}; rebuild on the tip first"),
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 pub fn create_project(&mut self, by: &Actor, name: &str, public: bool) -> Result<()> {
165 if !objects::valid_name(name) || name.contains(char::is_whitespace) {
166 return refused(format!("{name:?} can't be a project name"));
167 }
168 self.write(by, Self::rec("project", vec![("name", name.into()), ("public", public.into())]))
169 }
170
171 /// Record a point directly on main: a project's first snapshot, or imported history.
172 pub fn record_point(&mut self, by: &Actor, owner: &str, name: &str, snapshot: &Id, message: &str, author: Option<&Actor>) -> Result<u64> {
173 self.require_tree(snapshot)?;
174 let n = self.project(owner, name)?.tip() + 1;
175 let mut fields = vec![
176 ("owner", owner.into()),
177 ("project", name.into()),
178 ("n", n.into()),
179 ("snapshot", id_to_json(snapshot)),
180 ("message", message.into()),
181 ];
182 if let Some(a) = author {
183 fields.push(("author", a.to_json()));
184 }
185 self.write(by, Self::rec("point", fields))?;
186 Ok(n)
187 }
188
189 // ── Changes ────────────────────────────────────────────
190
191 /// Open a change as an idea: an intent, no code yet. Its first revision makes it a draft.
192 pub fn open_change(&mut self, by: &Actor, owner: &str, name: &str, intent: &str) -> Result<u64> {
193 let p = self.project(owner, name)?;
194 if !p.public && by.person != p.owner {
195 return refused(format!("{owner}/{name} is private"));
196 }
197 let number = p.changes.len() as u64 + 1;
198 self.write(by, Self::rec("change", vec![("owner", owner.into()), ("project", name.into()), ("number", number.into()), ("intent", intent.into()), ("state", "idea".into())]))?;
199 Ok(number)
200 }
201
202 /// Add a revision: the whole stack of steps, built on main point `base`.
203 pub fn revise(&mut self, by: &Actor, owner: &str, name: &str, change: u64, base: u64, steps: &[Step], note: Option<&str>) -> Result<u32> {
204 for s in steps {
205 self.require_tree(&s.snapshot)?;
206 }
207 let p = self.project(owner, name)?;
208 let c = p.change(change).ok_or_else(|| Error::Refused(format!("no change #{change}")))?;
209 if by.person != c.proposer.person && by.person != p.owner {
210 return refused("only the proposer or the owner can revise a change");
211 }
212 let number = c.revisions.len() as u32 + 1;
213 let steps: Vec<Value> = steps.iter().map(step_to_json).collect();
214 self.write(
215 by,
216 Self::rec("revision", vec![
217 ("owner", owner.into()),
218 ("project", name.into()),
219 ("change", change.into()),
220 ("number", (number as u64).into()),
221 ("base", base.into()),
222 ("steps", steps.into()),
223 ("note", note.map(str::to_string).into()),
224 ]),
225 )?;
226 Ok(number)
227 }
228
229 pub fn set_state(&mut self, by: &Actor, owner: &str, name: &str, change: u64, to: ChangeState) -> Result<()> {
230 if to == ChangeState::Landed {
231 return refused("changes land through land()");
232 }
233 self.write(by, Self::rec("state", vec![("owner", owner.into()), ("project", name.into()), ("change", change.into()), ("state", to.name().into())]))
234 }
235
236 /// Land a change's latest revision. Its steps become the next points on main, all at once.
237 pub fn land(&mut self, by: &Actor, owner: &str, name: &str, change: u64) -> Result<(u64, u64)> {
238 let p = self.project(owner, name)?;
239 if by.person != p.owner || by.agent.is_some() {
240 return refused("only the owner, in person, lands changes");
241 }
242 let c = p.change(change).ok_or_else(|| Error::Refused(format!("no change #{change}")))?;
243 let rev = c.latest().ok_or_else(|| Error::Refused(format!("#{change} has no revisions")))?;
244 if rev.base != p.tip() {
245 return Err(Error::Behind { base: rev.base, tip: p.tip() });
246 }
247 let first = p.tip() + 1;
248 let count = rev.steps.len() as u64;
249 let snapshots: Vec<Value> = rev.steps.iter().map(|s| id_to_json(&s.snapshot)).collect();
250 let revision = rev.number as u64;
251 self.write(
252 by,
253 Self::rec("land", vec![
254 ("owner", owner.into()),
255 ("project", name.into()),
256 ("change", change.into()),
257 ("revision", revision.into()),
258 ("first", first.into()),
259 ("snapshots", snapshots.into()),
260 ]),
261 )?;
262 Ok((first, first + count - 1))
263 }
264
265 pub fn say(&mut self, by: &Actor, owner: &str, name: &str, change: u64, body: &str) -> Result<()> {
266 self.write(by, Self::rec("say", vec![("owner", owner.into()), ("project", name.into()), ("change", change.into()), ("body", body.into())]))
267 }
268
269 /// Attach evidence to a revision: stored bytes of a media type, or a URL.
270 pub fn attach(&mut self, by: &Actor, owner: &str, name: &str, change: u64, revision: u32, title: &str, media_type: &str, content: Content) -> Result<()> {
271 let (blob, url) = match content {
272 Content::Bytes(bytes) => (Value::from(crate::hex::encode(&self.objects.put_blob(bytes)?)), Value::Null),
273 Content::Url(url) => (Value::Null, Value::from(url)),
274 };
275 self.write(
276 by,
277 Self::rec("evidence", vec![
278 ("owner", owner.into()),
279 ("project", name.into()),
280 ("change", change.into()),
281 ("revision", (revision as u64).into()),
282 ("title", title.into()),
283 ("type", media_type.into()),
284 ("blob", blob),
285 ("url", url),
286 ]),
287 )
288 }
289
290 /// Check that every object the state refers to exists and matches its id.
291 pub fn verify(&self) -> Result<usize> {
292 let mut count = 0;
293 let mut seen = std::collections::HashSet::new();
294 let mut check_tree = |id: &Id, count: &mut usize| -> Result<()> {
295 if seen.insert(*id) {
296 *count += self.objects.verify_tree(id)?;
297 }
298 Ok(())
299 };
300 for p in &self.state.projects {
301 for pt in &p.points {
302 check_tree(&pt.snapshot, &mut count)?;
303 }
304 for c in &p.changes {
305 for r in &c.revisions {
306 for s in &r.steps {
307 check_tree(&s.snapshot, &mut count)?;
308 }
309 }
310 for e in &c.evidence {
311 if let Some(b) = &e.blob {
312 self.objects.get_blob(b)?;
313 count += 1;
314 }
315 }
316 }
317 }
318 Ok(count)
319 }
320}
321
322#[cfg(test)]
323mod tests {
324 use super::*;
325 use crate::objects::{Entry, Kind};
326 use crate::testdir::TestDir;
327
328 fn tree(store: &Store, files: &[(&str, &str)]) -> Id {
329 let entries: Vec<Entry> = files
330 .iter()
331 .map(|(n, c)| Entry { name: n.to_string(), kind: Kind::File, id: store.objects.put_blob(c.as_bytes()).unwrap() })
332 .collect();
333 store.objects.put_tree(&entries).unwrap()
334 }
335
336 fn setup(dir: &TestDir) -> (Store, Actor, String) {
337 let mut store = Store::open(dir.path()).unwrap();
338 store.add_person("anton", "Anton").unwrap();
339 let agent = store.add_agent("anton", "claude").unwrap();
340 let me = Actor::agent("anton", &agent);
341 store.create_project(&Actor::person("anton"), "grove", true).unwrap();
342 let t = tree(&store, &[("README", "hello")]);
343 store.record_point(&Actor::person("anton"), "anton", "grove", &t, "Start", None).unwrap();
344 (store, me, agent)
345 }
346
347 #[test]
348 fn change_lifecycle_and_replay() {
349 let dir = TestDir::new("store-lifecycle");
350 {
351 let (mut store, me, _) = setup(&dir);
352 let owner = Actor::person("anton");
353 let c = store.open_change(&me, "anton", "grove", "Say more in the README").unwrap();
354 assert_eq!(store.project("anton", "grove").unwrap().change(c).unwrap().state, ChangeState::Idea);
355
356 let t1 = tree(&store, &[("README", "hello there")]);
357 let t2 = tree(&store, &[("README", "hello there"), ("NOTES", "x")]);
358 let steps = vec![
359 Step { id: "s1".into(), snapshot: t1, message: "Longer greeting".into() },
360 Step { id: "s2".into(), snapshot: t2, message: "Add notes".into() },
361 ];
362 assert_eq!(store.revise(&me, "anton", "grove", c, 1, &steps, None).unwrap(), 1);
363 store.set_state(&me, "anton", "grove", c, ChangeState::Proposed).unwrap();
364 store.say(&owner, "anton", "grove", c, "Looks good").unwrap();
365 store.attach(&me, "anton", "grove", c, 1, "notes", "text/plain", Content::Bytes(b"checked by hand")).unwrap();
366
367 // Only the owner lands.
368 assert!(matches!(store.land(&me, "anton", "grove", c), Err(Error::Refused(_))));
369 assert_eq!(store.land(&owner, "anton", "grove", c).unwrap(), (2, 3));
370 let p = store.project("anton", "grove").unwrap();
371 assert_eq!(p.tip(), 3);
372 assert_eq!(p.point(3).unwrap().snapshot, t2);
373 assert_eq!(p.change(c).unwrap().state, ChangeState::Landed);
374 assert_eq!(store.verify().unwrap() > 0, true);
375 }
376 // Everything comes back from the log.
377 let store = Store::open(dir.path()).unwrap();
378 let p = store.project("anton", "grove").unwrap();
379 assert_eq!(p.tip(), 3);
380 let c = p.change(1).unwrap();
381 assert_eq!(c.state, ChangeState::Landed);
382 assert_eq!(c.messages[0].body, "Looks good");
383 assert_eq!(c.evidence.len(), 1);
384 assert_eq!(p.point(2).unwrap().message, "Longer greeting");
385 assert!(p.point(2).unwrap().author.agent.is_some());
386 }
387
388 #[test]
389 fn landing_behind_the_tip_is_refused() {
390 let dir = TestDir::new("store-behind");
391 let (mut store, me, _) = setup(&dir);
392 let owner = Actor::person("anton");
393 let t = tree(&store, &[("README", "a")]);
394 let step = |id: &str| vec![Step { id: id.into(), snapshot: t, message: "m".into() }];
395 let a = store.open_change(&me, "anton", "grove", "a").unwrap();
396 let b = store.open_change(&me, "anton", "grove", "b").unwrap();
397 store.revise(&me, "anton", "grove", a, 1, &step("x"), None).unwrap();
398 store.revise(&me, "anton", "grove", b, 1, &step("y"), None).unwrap();
399 store.set_state(&me, "anton", "grove", a, ChangeState::Proposed).unwrap();
400 store.set_state(&me, "anton", "grove", b, ChangeState::Proposed).unwrap();
401 store.land(&owner, "anton", "grove", a).unwrap();
402 assert!(matches!(store.land(&owner, "anton", "grove", b), Err(Error::Behind { base: 1, tip: 2 })));
403 // Nothing changed for b.
404 assert_eq!(store.project("anton", "grove").unwrap().change(b).unwrap().state, ChangeState::Proposed);
405 }
406
407 #[test]
408 fn refusals_leave_no_trace() {
409 let dir = TestDir::new("store-refusals");
410 let (mut store, me, agent) = setup(&dir);
411 let before = store.state().records;
412 // Unstored snapshot, unknown project, bad transition, revoked agent.
413 let ghost = crate::objects::blob_id(b"not stored");
414 assert!(store.record_point(&Actor::person("anton"), "anton", "grove", &ghost, "x", None).is_err());
415 assert!(store.open_change(&me, "anton", "nope", "x").is_err());
416 let c = store.open_change(&me, "anton", "grove", "idea").unwrap();
417 assert!(store.set_state(&me, "anton", "grove", c, ChangeState::Proposed).is_err(), "an idea can't be proposed");
418 store.revoke_agent("anton", &agent).unwrap();
419 assert!(store.say(&me, "anton", "grove", c, "still here?").is_err());
420 assert_eq!(store.state().records, before + 2, "only the change and the revocation were written");
421 drop(store);
422 assert_eq!(Store::open(dir.path()).unwrap().state().records, before + 2);
423 }
424
425 #[test]
426 fn strangers_and_private_projects() {
427 let dir = TestDir::new("store-strangers");
428 let (mut store, _, _) = setup(&dir);
429 store.add_person("visitor", "Visitor").unwrap();
430 let visitor = Actor::person("visitor");
431 // Anyone can propose to a public project, but not land.
432 let c = store.open_change(&visitor, "anton", "grove", "An idea from outside").unwrap();
433 let t = tree(&store, &[("README", "outside")]);
434 store.revise(&visitor, "anton", "grove", c, 1, &[Step { id: "s".into(), snapshot: t, message: "m".into() }], None).unwrap();
435 store.set_state(&visitor, "anton", "grove", c, ChangeState::Proposed).unwrap();
436 assert!(store.land(&visitor, "anton", "grove", c).is_err());
437 store.set_state(&Actor::person("anton"), "anton", "grove", c, ChangeState::Declined).unwrap();
438 // Private projects take changes only from the owner.
439 store.create_project(&Actor::person("anton"), "secret", false).unwrap();
440 assert!(store.open_change(&visitor, "anton", "secret", "let me in").is_err());
441 }
442}