crates/nornir-saga/src/recorder.rs
//! **`SagaRecorder`** — the ergonomic *producer* emit-API a live app calls to
//! record its own UI/app journey.
//!
//! [`crate::record`] / [`crate::flush_saga_events`] are the process-global capture
//! buffer (one hidden `static` the whole binary shares) — perfect for a UI choke
//! that fires from anywhere, but awkward for an app that wants its OWN recorder
//! bound to a session and a destination. `SagaRecorder` is that instance-based
//! front door:
//!
//! ```no_run
//! use nornir_saga::{SagaRecorder, SessionCtx, MemTapeStore};
//! let ctx = SessionCtx::new("sess-1", "alice", "njord", "0.9.0", "abc123");
//! let rec = SagaRecorder::new(ctx, MemTapeStore::new());
//! rec.nav("/kund", "KundList", "open", r#"{"n":12}"#);
//! rec.click("/kund", "OffertForm", "submit", r#"{"ok":true}"#);
//! rec.flush().unwrap(); // durable → the store's warehouse/tape
//! ```
//!
//! It **buffers** [`SagaEvent`]s (stamping a monotonic `seq` + wall-clock `ts` and
//! the held [`SessionCtx`] identity, PII-redacted by default — the same discipline
//! as [`crate::record`]) and **flushes** them through a [`SagaTapeStore`]:
//!
//! * **fat** app → a warehouse-backed store (nornir's `WarehouseTapeStore` over the
//! live Mimir `saga_events` table);
//! * **thin** app → a gRPC-backed store that ships the batch to a nornir server's
//! saga ingest endpoint;
//! * **leaf** app / test → the built-in [`crate::FileTapeStore`] / [`MemTapeStore`].
//!
//! The recorder never names a backend — it holds any `S: SagaTapeStore`, so the
//! SAME producer code records fat or thin (REUSE: one client, every app consumes
//! it). Interior-mutable (`&self`) so it can live behind an `Arc` shared across an
//! app's callbacks.
//!
//! [`SagaTapeStore`]: crate::store::SagaTapeStore
//! [`MemTapeStore`]: crate::store::MemTapeStore
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::Result;
use crate::event::{SagaEvent, SessionCtx};
use crate::store::SagaTapeStore;
/// An instance-based saga recorder: buffer captured events (stamped with a
/// monotonic `seq`, a wall-clock `ts`, and the held session identity — PII-redacted
/// by default), then durably [`flush`](SagaRecorder::flush) them through any
/// [`SagaTapeStore`].
///
/// Generic over the store `S` so a producer picks its destination at construction
/// (warehouse / gRPC / file / memory) and the recording code stays identical.
pub struct SagaRecorder<S: SagaTapeStore> {
ctx: SessionCtx,
store: S,
buf: Mutex<Vec<SagaEvent>>,
seq: AtomicU64,
}
impl<S: SagaTapeStore> SagaRecorder<S> {
/// A recorder for `ctx`'s session, flushing to `store`. Capture `seq` starts at
/// `0` (use [`with_start_seq`](SagaRecorder::with_start_seq) to continue an
/// existing tape's numbering).
pub fn new(ctx: SessionCtx, store: S) -> Self {
Self::with_start_seq(ctx, store, 0)
}
/// A recorder whose first captured event is stamped `start_seq` (then
/// monotonically increasing) — for resuming a session whose earlier events
/// already landed, so replay order stays a single strictly-increasing run.
pub fn with_start_seq(ctx: SessionCtx, store: S, start_seq: u64) -> Self {
Self { ctx, store, buf: Mutex::new(Vec::new()), seq: AtomicU64::new(start_seq) }
}
/// The session identity every recorded event is stamped with.
pub fn ctx(&self) -> &SessionCtx {
&self.ctx
}
/// The destination store (to `list_sessions` / `read_session` the tape back).
pub fn store(&self) -> &S {
&self.store
}
/// Number of events buffered but not yet flushed.
pub fn pending(&self) -> usize {
self.buf.lock().unwrap_or_else(|p| p.into_inner()).len()
}
fn now_micros() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_micros() as u64).unwrap_or(0)
}
/// Stamp session identity + a monotonic `seq` + a wall-clock `ts` onto `event`
/// (PII-redacted by default) and buffer it. This is the general entry point;
/// the [`nav`](SagaRecorder::nav)/[`click`](SagaRecorder::click)/… helpers build
/// the event for you. Lock poisoning is recovered from — a panic mid-record must
/// never wedge an app's capture.
pub fn record(&self, event: SagaEvent) {
// Identity first (from the held ctx), then mask-by-default (same order the
// process-global `record` guarantees: a stored journey never carries a raw
// email/token), then stamp ordering.
let mut event = event.in_session(&self.ctx).redacted();
event.seq = self.seq.fetch_add(1, Ordering::Relaxed);
if event.ts_micros == 0 {
event.ts_micros = Self::now_micros();
}
self.buf.lock().unwrap_or_else(|p| p.into_inner()).push(event);
}
/// Record a **navigation** — the primary, replayable step.
pub fn nav(&self, route: &str, component: &str, action: &str, state_json: &str) {
self.record(SagaEvent::nav(route, component, action, state_json));
}
/// Record a **click** / activation — also replayable.
pub fn click(&self, route: &str, component: &str, action: &str, state_json: &str) {
self.record(SagaEvent::click(route, component, action, state_json));
}
/// Record a **log** line (context, not a replay step).
pub fn log(&self, level: &str, detail: &str) {
self.record(SagaEvent::log(level, detail));
}
/// Record an **error** (surfaced distinctly; drives the painful-first ranking).
pub fn error(&self, component: &str, detail: &str) {
self.record(SagaEvent::error(component, detail));
}
/// Record a standalone component-**state** snapshot.
pub fn state(&self, component: &str, state_json: &str) {
self.record(SagaEvent::state(component, state_json));
}
/// Durably flush the buffered events to the store, clearing the buffer **only
/// after** the append succeeds. Crash-safe: a failed append RETAINS the batch
/// (so the flush can be retried and no captured journey is lost to a transient
/// store error), exactly like [`crate::flush_saga_events`]. An empty buffer is a
/// no-op success. Returns the number of events flushed.
pub fn flush(&self) -> Result<usize> {
let mut buf = self.buf.lock().unwrap_or_else(|p| p.into_inner());
if buf.is_empty() {
return Ok(0);
}
self.store.append_events(&buf)?; // append FIRST, still holding the buffer
let n = buf.len();
buf.clear(); // clear ONLY once the batch is durable
Ok(n)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::store::MemTapeStore;
#[test]
fn records_stamps_seq_identity_and_flushes_through_the_store() {
let ctx = SessionCtx::new("s-rec", "alice", "njord", "0.9.0", "abc123");
let rec = SagaRecorder::new(ctx, MemTapeStore::new());
rec.nav("/kund", "KundList", "open", r#"{"n":12}"#);
rec.click("/kund", "OffertForm", "submit", r#"{"ok":true}"#);
assert_eq!(rec.pending(), 2, "two events buffered, not yet flushed");
// Nothing is in the store until flush.
assert!(rec.store().list_sessions().unwrap().is_empty(), "no session before flush");
let n = rec.flush().unwrap();
assert_eq!(n, 2, "both events flushed");
assert_eq!(rec.pending(), 0, "buffer cleared on a durable flush");
// The tape reads back scoped + seq-ordered, identity + state_json intact.
let back = rec.store().read_session("s-rec").unwrap();
assert_eq!(back.len(), 2);
assert!(back.iter().all(|e| e.session_id == "s-rec" && e.app == "njord"));
assert!(back[0].seq < back[1].seq, "monotonic seq in record order");
assert_eq!(back[0].component, "KundList");
assert_eq!(back[1].state_json, r#"{"ok":true}"#, "PII-free state round-trips exactly");
assert!(back[0].ts_micros > 0, "recorder stamps a wall-clock ts");
}
#[test]
fn flush_is_crash_safe_retaining_the_batch_on_a_store_error() {
struct FailStore;
impl SagaTapeStore for FailStore {
fn append_events(&self, _e: &[SagaEvent]) -> Result<()> {
Err(anyhow::anyhow!("store down"))
}
fn list_sessions(&self) -> Result<Vec<crate::analysis::SagaSessionSummary>> {
Ok(Vec::new())
}
fn read_session(&self, _s: &str) -> Result<Vec<SagaEvent>> {
Ok(Vec::new())
}
}
let ctx = SessionCtx::new("s", "u", "app", "1", "sha");
let rec = SagaRecorder::new(ctx, FailStore);
rec.nav("/a", "A", "open", "{}");
assert!(rec.flush().is_err(), "the store error surfaces");
assert_eq!(rec.pending(), 1, "a failed flush RETAINS the batch (no loss)");
}
#[test]
fn redacts_client_pii_by_default() {
let ctx = SessionCtx::new("s-pii", "u", "app", "1", "sha");
let rec = SagaRecorder::new(ctx, MemTapeStore::new());
rec.nav("/a", "A", "open", r#"{"email":"bob@example.com"}"#);
rec.flush().unwrap();
let back = rec.store().read_session("s-pii").unwrap();
assert!(
back[0].state_json.contains(crate::event::redact::MASK),
"recorder masks PII by default, got {:?}",
back[0].state_json
);
}
#[test]
fn with_start_seq_continues_a_tape_numbering() {
let ctx = SessionCtx::new("s2", "u", "app", "1", "sha");
let rec = SagaRecorder::with_start_seq(ctx, MemTapeStore::new(), 100);
rec.nav("/a", "A", "open", "{}");
rec.nav("/b", "B", "open", "{}");
rec.flush().unwrap();
let back = rec.store().read_session("s2").unwrap();
assert_eq!(back[0].seq, 100, "first event resumes at start_seq");
assert_eq!(back[1].seq, 101, "then monotonically increases");
}
}