ocel_transform/
apply.rs

1//! Applying a recipe: each step maps a valid log to a valid log.
2//!
3//! Event-level steps run through [`ocel_etl::StagingLog`] (whose gate
4//! re-validates), object-level steps use the core's OCEL-aware filters, and
5//! the final result is validated once more — a recipe can drop data, but it
6//! can never produce an inconsistent log.
7
8use std::collections::BTreeSet;
9
10use chrono::{DateTime, NaiveDate, Utc};
11use ocel::{AttrValue, Ocel, Violation};
12use ocel_etl::StagingLog;
13use regex::Regex;
14use thiserror::Error;
15
16use crate::recipe::{EventPredicate, Recipe, Step};
17
18#[derive(Debug, Error)]
19pub enum TransformError {
20    #[error("step {step}: invalid regex: {message}")]
21    BadRegex { step: usize, message: String },
22
23    #[error("step {step}: invalid time bound {value:?} (RFC 3339 or YYYY-MM-DD)")]
24    BadTime { step: usize, value: String },
25
26    #[error("step {step}: predicate has no conditions (it would drop every event)")]
27    EmptyPredicate { step: usize },
28
29    #[error("step {step}: value conditions (equals/matches/min/max) require `attr`")]
30    ValueConditionWithoutAttr { step: usize },
31
32    #[error("the transformed log failed validation: {0:?}")]
33    Invalid(Vec<Violation>),
34}
35
36/// Effect of one step, for honest reporting in CLIs and UIs.
37#[derive(Debug, Clone, serde::Serialize)]
38#[serde(rename_all = "camelCase")]
39pub struct StepReport {
40    pub step: String,
41    pub events_before: usize,
42    pub events_after: usize,
43    pub objects_before: usize,
44    pub objects_after: usize,
45}
46
47/// Apply every step of `recipe` in order.
48pub fn apply(recipe: &Recipe, log: Ocel) -> Result<(Ocel, Vec<StepReport>), TransformError> {
49    let mut log = log;
50    let mut reports = Vec::with_capacity(recipe.steps.len());
51    for (index, step) in recipe.steps.iter().enumerate() {
52        let events_before = log.events.len();
53        let objects_before = log.objects.len();
54        log = apply_step(index, step, log)?;
55        reports.push(StepReport {
56            step: step.label().to_owned(),
57            events_before,
58            events_after: log.events.len(),
59            objects_before,
60            objects_after: log.objects.len(),
61        });
62    }
63    log.validate().map_err(TransformError::Invalid)?;
64    Ok((log, reports))
65}
66
67/// A dropped event, summarized for a "this is what the recipe deletes"
68/// preview — machine cleaning must stay human-inspectable.
69#[derive(Debug, Clone, serde::Serialize)]
70#[serde(rename_all = "camelCase")]
71pub struct DroppedEvent {
72    pub id: String,
73    pub event_type: String,
74    pub time: DateTime<Utc>,
75    /// (name, value as text) pairs.
76    pub attributes: Vec<(String, String)>,
77}
78
79/// One step's effect plus a sample of the events it removed.
80#[derive(Debug, Clone, serde::Serialize)]
81#[serde(rename_all = "camelCase")]
82pub struct StepPreview {
83    #[serde(flatten)]
84    pub report: StepReport,
85    /// Up to `sample` of the events this step dropped.
86    pub dropped_events: Vec<DroppedEvent>,
87    /// Total events this step dropped (the sample may be shorter).
88    pub dropped_total: usize,
89}
90
91/// Like [`apply`], but additionally samples the events each step drops.
92/// Costs one id-set pass per step; meant for interactive previews.
93pub fn preview(
94    recipe: &Recipe,
95    log: Ocel,
96    sample: usize,
97) -> Result<(Ocel, Vec<StepPreview>), TransformError> {
98    let mut log = log;
99    let mut previews = Vec::with_capacity(recipe.steps.len());
100    for (index, step) in recipe.steps.iter().enumerate() {
101        let before = log.clone();
102        log = apply_step(index, step, log)?;
103        let after_ids: BTreeSet<&str> = log.events.iter().map(|e| e.id.as_str()).collect();
104        let dropped: Vec<&ocel::Event> = before
105            .events
106            .iter()
107            .filter(|e| !after_ids.contains(e.id.as_str()))
108            .collect();
109        let dropped_events = dropped
110            .iter()
111            .take(sample)
112            .map(|e| DroppedEvent {
113                id: e.id.clone(),
114                event_type: e.event_type.clone(),
115                time: e.time,
116                attributes: e
117                    .attributes
118                    .iter()
119                    .map(|a| (a.name.clone(), a.value.to_text()))
120                    .collect(),
121            })
122            .collect();
123        previews.push(StepPreview {
124            report: StepReport {
125                step: step.label().to_owned(),
126                events_before: before.events.len(),
127                events_after: log.events.len(),
128                objects_before: before.objects.len(),
129                objects_after: log.objects.len(),
130            },
131            dropped_events,
132            dropped_total: dropped.len(),
133        });
134    }
135    log.validate().map_err(TransformError::Invalid)?;
136    Ok((log, previews))
137}
138
139fn apply_step(index: usize, step: &Step, log: Ocel) -> Result<Ocel, TransformError> {
140    match step {
141        Step::DropEventTypes(names) => Ok(with_staging(log, |staging| {
142            staging.retain_events(|e| !names.contains(&e.event_type));
143        })),
144        Step::KeepEventTypes(names) => Ok(with_staging(log, |staging| {
145            staging.retain_events(|e| names.contains(&e.event_type));
146        })),
147        Step::DropEventsWhere(predicate) => {
148            let matcher = Matcher::compile(index, predicate)?;
149            Ok(with_staging(log, |staging| {
150                staging.retain_events(|e| !matcher.matches(&e.event_type, &e.attributes));
151            }))
152        }
153        Step::RenameEventTypes(renames) => Ok(with_staging(log, |staging| {
154            staging.map_events(|e| {
155                if let Some(new_name) = renames.get(&e.event_type) {
156                    e.event_type.clone_from(new_name);
157                }
158            });
159        })),
160        Step::TimeWindow(window) => {
161            let from = parse_bound(index, window.from.as_deref(), false)?;
162            let to = parse_bound(index, window.to.as_deref(), true)?;
163            Ok(with_staging(log, |staging| {
164                staging.retain_events(|e| {
165                    from.is_none_or(|f| e.time >= f) && to.is_none_or(|t| e.time < t)
166                });
167            }))
168        }
169        Step::KeepObjectTypes(names) => {
170            let names: Vec<&str> = names.iter().map(String::as_str).collect();
171            Ok(log.filter_object_types(&names))
172        }
173        Step::DropObjectsWithoutEvents => Ok(drop_objects_without_events(log)),
174    }
175}
176
177/// Round-trip through the staging representation; the gate cannot fail here
178/// because the input was valid and these transforms only drop or rename.
179fn with_staging(log: Ocel, f: impl FnOnce(&mut StagingLog)) -> Ocel {
180    let mut staging = StagingLog::from_ocel(log);
181    f(&mut staging);
182    staging
183        .into_ocel()
184        .expect("dropping or renaming events keeps a valid log valid")
185}
186
187fn drop_objects_without_events(log: Ocel) -> Ocel {
188    let referenced: BTreeSet<String> = log
189        .events
190        .iter()
191        .flat_map(|e| e.relationships.iter().map(|r| r.object_id.clone()))
192        .collect();
193    let objects = log
194        .objects
195        .into_iter()
196        .filter(|o| referenced.contains(&o.id))
197        .map(|mut o| {
198            o.relationships
199                .retain(|r| referenced.contains(&r.object_id));
200            o
201        })
202        .collect();
203    Ocel { objects, ..log }
204}
205
206/// A compiled event predicate.
207struct Matcher {
208    event_type: Option<String>,
209    attr: Option<String>,
210    equals: Option<String>,
211    regex: Option<Regex>,
212    min: Option<f64>,
213    max: Option<f64>,
214}
215
216impl Matcher {
217    fn compile(step: usize, p: &EventPredicate) -> Result<Matcher, TransformError> {
218        let has_value_condition =
219            p.equals.is_some() || p.matches.is_some() || p.min.is_some() || p.max.is_some();
220        if p.event_type.is_none() && p.attr.is_none() {
221            return Err(TransformError::EmptyPredicate { step });
222        }
223        if has_value_condition && p.attr.is_none() {
224            return Err(TransformError::ValueConditionWithoutAttr { step });
225        }
226        let regex = p
227            .matches
228            .as_deref()
229            .map(Regex::new)
230            .transpose()
231            .map_err(|e| TransformError::BadRegex {
232                step,
233                message: e.to_string(),
234            })?;
235        Ok(Matcher {
236            event_type: p.event_type.clone(),
237            attr: p.attr.clone(),
238            equals: p.equals.clone(),
239            regex,
240            min: p.min,
241            max: p.max,
242        })
243    }
244
245    fn matches(&self, event_type: &str, attributes: &[(String, AttrValue)]) -> bool {
246        if let Some(ty) = &self.event_type {
247            if event_type != ty {
248                return false;
249            }
250        }
251        let Some(attr) = &self.attr else {
252            return true; // type-only predicate
253        };
254        let Some((_, value)) = attributes.iter().find(|(name, _)| name == attr) else {
255            return false; // events without the attribute never match
256        };
257        let text = value.to_text();
258        if let Some(equals) = &self.equals {
259            if text != *equals {
260                return false;
261            }
262        }
263        if let Some(regex) = &self.regex {
264            if !regex.is_match(&text) {
265                return false;
266            }
267        }
268        if self.min.is_some() || self.max.is_some() {
269            // numeric conditions apply to numeric values only; intentional cast
270            #[allow(clippy::cast_precision_loss)]
271            let number = match value {
272                AttrValue::Integer(i) => Some(*i as f64),
273                AttrValue::Float(f) => Some(*f),
274                AttrValue::String(_) | AttrValue::Boolean(_) | AttrValue::Time(_) => None,
275            };
276            let Some(number) = number else {
277                return false;
278            };
279            if self.min.is_some_and(|min| number < min) {
280                return false;
281            }
282            if self.max.is_some_and(|max| number > max) {
283                return false;
284            }
285        }
286        true
287    }
288}
289
290/// RFC 3339, or a date: `from` dates mean that midnight, `to` dates mean the
291/// *following* midnight so the named day is included in the half-open window.
292fn parse_bound(
293    step: usize,
294    bound: Option<&str>,
295    is_to: bool,
296) -> Result<Option<DateTime<Utc>>, TransformError> {
297    let Some(raw) = bound else {
298        return Ok(None);
299    };
300    if let Ok(instant) = DateTime::parse_from_rfc3339(raw) {
301        return Ok(Some(instant.to_utc()));
302    }
303    let Ok(date) = NaiveDate::parse_from_str(raw, "%Y-%m-%d") else {
304        return Err(TransformError::BadTime {
305            step,
306            value: raw.to_owned(),
307        });
308    };
309    let date = if is_to { date.succ_opt() } else { Some(date) };
310    let midnight =
311        date.and_then(|d| d.and_hms_opt(0, 0, 0))
312            .ok_or_else(|| TransformError::BadTime {
313                step,
314                value: raw.to_owned(),
315            })?;
316    Ok(Some(midnight.and_utc()))
317}
ocel_transform/apply.rs

ocel_transform/
apply.rs
