levi_core/
materialize.rs

1//! Deterministic replay of the levi event log into materialized state.
2//!
3//! myko applies events in arrival order; levi guarantees the spec's LWW
4//! ordering — `created_at`, ties broken by event id — by sorting every record
5//! with that key before replay. Event id = git blob OID of the event's CBOR
6//! bytes, so ordering is identical on every node (spec: "Clock skew").
7
8use std::collections::BTreeMap;
9
10use chrono::{DateTime, Duration, Utc};
11use myko::prelude::Eventable;
12use myko::wire::{MEvent, MEventType};
13
14use crate::entities::*;
15
16/// One event as stored in the ref: `id` is the git blob OID (hex) of the CBOR
17/// bytes, `event` the decoded myko wire event.
18#[derive(Debug, Clone)]
19pub struct EventRecord {
20    pub id: String,
21    pub event: MEvent,
22}
23
24/// Materialized state of one project's event log.
25#[derive(Debug, Default)]
26pub struct World {
27    pub project: Option<Project>,
28    pub tasks: BTreeMap<String, Task>,
29    /// Sorted by (at, id).
30    pub status_changes: Vec<StatusChange>,
31    pub deps: BTreeMap<String, Dependency>,
32    /// Keyed by task_id (== Claim entity id); SET overwrite = newest wins.
33    pub claims: BTreeMap<String, Claim>,
34    /// Sorted by (at, id).
35    pub comments: Vec<Comment>,
36    pub commit_facts: BTreeMap<String, CommitFact>,
37    pub ref_facts: BTreeMap<String, RefFact>,
38}
39
40impl World {
41    /// This task's status changes, in (at, id) order.
42    pub fn changes_for(&self, task_id: &str) -> Vec<&StatusChange> {
43        self.status_changes
44            .iter()
45            .filter(|c| &*c.task_id == task_id)
46            .collect()
47    }
48
49    /// The claim on this task, if it exists and its ttl has not expired.
50    pub fn live_claim(&self, task_id: &str, now: DateTime<Utc>) -> Option<&Claim> {
51        let claim = self.claims.get(task_id)?;
52        let at = DateTime::parse_from_rfc3339(&claim.created)
53            .ok()?
54            .with_timezone(&Utc);
55        let ttl = Duration::seconds(i64::try_from(claim.ttl_secs).ok()?);
56        (at + ttl > now).then_some(claim)
57    }
58}
59
60pub fn materialize(mut records: Vec<EventRecord>) -> World {
61    records.sort_by(|a, b| {
62        (a.event.created_at.as_str(), a.id.as_str())
63            .cmp(&(b.event.created_at.as_str(), b.id.as_str()))
64    });
65
66    let mut w = World::default();
67    let mut status_changes: BTreeMap<String, StatusChange> = BTreeMap::new();
68    let mut comments: BTreeMap<String, Comment> = BTreeMap::new();
69
70    for r in &records {
71        let ev = &r.event;
72        let t = ev.item_type.as_str();
73        if t == Project::ENTITY_NAME_STATIC {
74            if matches!(ev.change_type, MEventType::SET)
75                && let Ok(p) = serde_json::from_value::<Project>(ev.item.clone())
76            {
77                w.project = Some(p);
78            }
79        } else if t == Task::ENTITY_NAME_STATIC {
80            apply(&mut w.tasks, ev);
81        } else if t == StatusChange::ENTITY_NAME_STATIC {
82            apply(&mut status_changes, ev);
83        } else if t == Dependency::ENTITY_NAME_STATIC {
84            apply(&mut w.deps, ev);
85        } else if t == Claim::ENTITY_NAME_STATIC {
86            apply(&mut w.claims, ev);
87        } else if t == Comment::ENTITY_NAME_STATIC {
88            apply(&mut comments, ev);
89        } else if t == CommitFact::ENTITY_NAME_STATIC {
90            apply(&mut w.commit_facts, ev);
91        } else if t == RefFact::ENTITY_NAME_STATIC {
92            apply(&mut w.ref_facts, ev);
93        }
94        // Unknown item types are skipped: forward compatibility with newer levi.
95    }
96
97    w.status_changes = status_changes.into_values().collect();
98    w.status_changes
99        .sort_by(|a, b| (a.created.as_str(), &*a.id.0).cmp(&(b.created.as_str(), &*b.id.0)));
100    w.comments = comments.into_values().collect();
101    w.comments
102        .sort_by(|a, b| (a.created.as_str(), &*a.id.0).cmp(&(b.created.as_str(), &*b.id.0)));
103    w
104}
105
106/// SET upserts by entity id; DEL removes (tombstone applies even when the
107/// item was never seen — events replay in deterministic order, so this
108/// converges on every node).
109fn apply<T>(map: &mut BTreeMap<String, T>, ev: &MEvent)
110where
111    T: Eventable,
112{
113    match ev.change_type {
114        MEventType::SET => {
115            if let Ok(item) = serde_json::from_value::<T>(ev.item.clone()) {
116                map.insert(item.id().to_string(), item);
117            }
118        }
119        MEventType::DEL => {
120            if let Some(id) = ev.item.get("id").and_then(|v| v.as_str()) {
121                map.remove(id);
122            }
123        }
124    }
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130
131    fn rec<T: myko::prelude::Eventable>(oid: &str, item: &T, created_at: &str) -> EventRecord {
132        let mut event = MEvent::from_item(item, MEventType::SET, "m");
133        event.created_at = created_at.to_string();
134        EventRecord {
135            id: oid.to_string(),
136            event,
137        }
138    }
139
140    fn del<T: myko::prelude::Eventable>(oid: &str, item: &T, created_at: &str) -> EventRecord {
141        let mut event = MEvent::from_item(item, MEventType::DEL, "m");
142        event.created_at = created_at.to_string();
143        EventRecord {
144            id: oid.to_string(),
145            event,
146        }
147    }
148
149    fn task(id: &str, title: &str) -> Task {
150        Task {
151            id: id.into(),
152            project_id: "p".into(),
153            title: title.into(),
154            body: String::new(),
155            priority: Priority::default(),
156            labels: vec![],
157            created_by_dev: "d".into(),
158            created_by_machine: "m".into(),
159            created: "2026-07-01T00:00:00Z".into(),
160        }
161    }
162
163    #[test]
164    fn lww_by_created_at_regardless_of_arrival_order() {
165        // The later edit arrives first in the vec; sorting must fix it.
166        let w = materialize(vec![
167            rec("bb", &task("t1", "newer title"), "2026-07-02T00:00:00Z"),
168            rec("aa", &task("t1", "older title"), "2026-07-01T00:00:00Z"),
169        ]);
170        assert_eq!(w.tasks["t1"].title, "newer title");
171    }
172
173    #[test]
174    fn created_at_tie_broken_by_event_id() {
175        let at = "2026-07-01T00:00:00Z";
176        let w = materialize(vec![
177            rec("ff", &task("t1", "id ff wins"), at),
178            rec("aa", &task("t1", "id aa loses"), at),
179        ]);
180        assert_eq!(w.tasks["t1"].title, "id ff wins");
181    }
182
183    #[test]
184    fn del_removes_dependency() {
185        let d = Dependency {
186            id: dependency_id("a", "b").into(),
187            project_id: "p".into(),
188            blocker_task_id: "a".into(),
189            blocked_task_id: "b".into(),
190            blocker_project_id: None,
191            blocker_ref: None,
192            via: None,
193        };
194        let w = materialize(vec![
195            rec("aa", &d, "2026-07-01T00:00:00Z"),
196            del("bb", &d, "2026-07-02T00:00:00Z"),
197        ]);
198        assert!(w.deps.is_empty());
199    }
200
201    #[test]
202    fn claim_liveness_respects_ttl() {
203        let c = Claim {
204            id: "t1".into(),
205            project_id: "p".into(),
206            task_id: "t1".into(),
207            dev: "d".into(),
208            machine: "m".into(),
209            worktree: "/w".into(),
210            machine_id: String::new(),
211            created: "2026-07-01T00:00:00Z".into(),
212            ttl_secs: 3600,
213        };
214        let w = materialize(vec![rec("aa", &c, "2026-07-01T00:00:00Z")]);
215        let live = "2026-07-01T00:30:00Z".parse::<DateTime<Utc>>().unwrap();
216        let dead = "2026-07-01T02:00:00Z".parse::<DateTime<Utc>>().unwrap();
217        assert!(w.live_claim("t1", live).is_some());
218        assert!(w.live_claim("t1", dead).is_none());
219    }
220
221    #[test]
222    fn project_and_sorted_append_only_sets() {
223        let p = Project {
224            id: "p".into(),
225            name: "levi".into(),
226            created: "2026-07-01T00:00:00Z".into(),
227        };
228        let c2 = Comment {
229            id: "c2".into(),
230            project_id: "p".into(),
231            task_id: "t1".into(),
232            body: "second".into(),
233            by_dev: "d".into(),
234            created: "2026-07-02T00:00:00Z".into(),
235        };
236        let c1 = Comment {
237            id: "c1".into(),
238            project_id: "p".into(),
239            task_id: "t1".into(),
240            body: "first".into(),
241            by_dev: "d".into(),
242            created: "2026-07-01T00:00:00Z".into(),
243        };
244        let w = materialize(vec![
245            rec("cc", &c2, "2026-07-02T00:00:00Z"),
246            rec("aa", &p, "2026-07-01T00:00:00Z"),
247            rec("bb", &c1, "2026-07-01T00:00:00Z"),
248        ]);
249        assert_eq!(w.project.as_ref().unwrap().name, "levi");
250        let bodies: Vec<_> = w.comments.iter().map(|c| c.body.as_str()).collect();
251        assert_eq!(bodies, ["first", "second"]);
252    }
253
254    #[test]
255    fn unknown_item_type_is_ignored() {
256        let mut event = MEvent::from_item(&task("t1", "x"), MEventType::SET, "m");
257        event.item_type = "FutureThing".into();
258        let w = materialize(vec![EventRecord {
259            id: "aa".into(),
260            event,
261        }]);
262        assert!(w.tasks.is_empty());
263    }
264}
levi_core/materialize.rs

levi_core/
materialize.rs
