levi_core/
hub.rs

1//! Hub-side machinery: unwrap incoming LogEntries into real entities so
2//! aggregate queries and the dashboard see Tasks/StatusChanges/etc.
3//! (spec §Sync, hub leg).
4//!
5//! Note this command never inspects entity types itself — `emit_event_batch`
6//! routes by `item_type` exactly like direct ingestion. The command exists
7//! because the hub keeps the *journal* (LogEntries carry the original
8//! content-addressed bytes the pull leg needs), and because it is the
9//! interposition point for LWW ordering: myko applies raw events in arrival
10//! order, so without the [`Applied`] guard an event synced late would
11//! clobber newer state that every CLI (which sorts the full log before
12//! replay) resolves correctly.
13
14use std::sync::Arc;
15
16use myko::command::{CommandContext, CommandError, CommandHandler};
17use myko::prelude::*;
18use myko::saga::{SagaContext, SagaHandler};
19use myko::wire::MEventType;
20
21use crate::entities::{Applied, GetAppliedsByIds, LogEntry, applied_id};
22
23/// Apply the event wrapped inside a LogEntry to this node's stores, unless a
24/// newer event for the same entity was already applied (LWW by
25/// (created_at, event id), the same key `materialize()` sorts by).
26/// Idempotent: re-applying converges.
27#[myko_command]
28pub struct ApplyLogEntry {
29    pub cbor_b64: String,
30    /// The LogEntry id — the event's content address (git blob OID).
31    pub event_oid: String,
32}
33
34impl CommandHandler for ApplyLogEntry {
35    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
36        use base64::Engine as _;
37        let err = |m: String| CommandError {
38            tx: ctx.tx().to_string(),
39            command_id: ctx.command_id.to_string(),
40            message: m,
41        };
42        let bytes = base64::engine::general_purpose::STANDARD
43            .decode(&self.cbor_b64)
44            .map_err(|e| err(format!("invalid base64 in LogEntry: {e}")))?;
45        let event: myko::wire::MEvent = ciborium::from_reader(bytes.as_slice())
46            .map_err(|e| err(format!("invalid CBOR in LogEntry: {e}")))?;
47        // Never let a wrapped LogEntry apply another LogEntry: no recursion.
48        if event.item_type == LogEntry::ENTITY_NAME_STATIC {
49            return Err(err("refusing to unwrap a nested LogEntry".into()));
50        }
51        let Some(item_id) = event.item.get("id").and_then(|v| v.as_str()) else {
52            return Err(err("wrapped event has no item id".into()));
53        };
54
55        // LWW guard: skip events older than what this entity already shows.
56        let guard_id = applied_id(&event.item_type, item_id);
57        let existing = ctx.exec_query_first(GetAppliedsByIds {
58            ids: vec![guard_id.clone().into()],
59        })?;
60        let incoming = (event.created_at.clone(), self.event_oid.clone());
61        if let Some(applied) = existing
62            && (applied.event_created.as_str(), applied.event_oid.as_str())
63                > (incoming.0.as_str(), incoming.1.as_str())
64        {
65            return Ok(()); // stale: a newer event was already applied
66        }
67
68        ctx.emit_event_batch(vec![event])?;
69        ctx.emit_set(&Applied {
70            id: guard_id.into(),
71            event_created: incoming.0,
72            event_oid: incoming.1,
73        })?;
74        Ok(())
75    }
76}
77
78myko::register_command_handler!(ApplyLogEntry);
79
80/// Saga: every LogEntry SET (from CLI pushes or Postgres replay) is unwrapped
81/// into its inner event.
82#[myko_saga]
83pub struct UnwrapLogEntries;
84
85impl SagaHandler for UnwrapLogEntries {
86    type EventItem = LogEntry;
87    type Command = ApplyLogEntry;
88    const EVENT_TYPE: MEventType = MEventType::SET;
89
90    fn handle(
91        item: LogEntry,
92        _event: myko::wire::MEvent,
93        _ctx: Arc<SagaContext>,
94    ) -> Option<ApplyLogEntry> {
95        Some(ApplyLogEntry {
96            cbor_b64: item.cbor_b64,
97            event_oid: item.id.to_string(),
98        })
99    }
100}
levi_core/hub.rs

levi_core/
hub.rs
