teaql_runtime/
event.rs

1use std::sync::Arc;
2
3use teaql_core::{Record, Value};
4
5use crate::{RuntimeError, UserContext};
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum RawAuditEventKind {
9    Created,
10    Updated,
11    Deleted,
12    Recovered,
13    /// Emitted when a new table is created during schema bootstrap.
14    SchemaCreated,
15    /// Emitted when an existing table is verified during schema bootstrap.
16    SchemaVerified,
17    /// Emitted when a new column is added to an existing table (schema migration).
18    FieldAdded,
19    /// Emitted when initial seed data is inserted or updated during bootstrap.
20    DataSeeded,
21}
22
23#[derive(Debug, Clone, PartialEq)]
24pub struct EntityPropertyChange {
25    pub field: String,
26    pub old_value: Option<Value>,
27    pub new_value: Option<Value>,
28}
29
30impl EntityPropertyChange {
31    pub fn new(
32        field: impl Into<String>,
33        old_value: Option<Value>,
34        new_value: Option<Value>,
35    ) -> Self {
36        Self {
37            field: field.into(),
38            old_value,
39            new_value,
40        }
41    }
42}
43
44#[derive(Debug, Clone, PartialEq)]
45pub struct RawAuditEvent {
46    pub kind: RawAuditEventKind,
47    pub entity: String,
48    pub values: Record,
49    pub updated_fields: Vec<String>,
50    pub old_values: Option<Record>,
51    pub new_values: Option<Record>,
52    pub changes: Vec<EntityPropertyChange>,
53    /// Annotation trace chain from the graph save scope chain.
54    pub trace_chain: Vec<teaql_core::TraceNode>,
55    pub bootstrap_audit: Option<BootstrapAuditIdentity>,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct BootstrapAuditIdentity {
60    pub actor: String,
61    pub category: String,
62    pub reason: String,
63    pub resulting_version: Option<i64>,
64    pub occurred_at_millis: u64,
65}
66
67impl RawAuditEvent {
68    pub fn created(entity: impl Into<String>, values: Record) -> Self {
69        let changes = values
70            .iter()
71            .map(|(field, value)| {
72                EntityPropertyChange::new(field.clone(), None, Some(value.clone()))
73            })
74            .collect();
75        Self {
76            kind: RawAuditEventKind::Created,
77            entity: entity.into(),
78            values: values.clone(),
79            updated_fields: Vec::new(),
80            old_values: None,
81            new_values: Some(values),
82            changes,
83            trace_chain: Vec::new(),
84            bootstrap_audit: None,
85        }
86    }
87
88    pub fn updated(entity: impl Into<String>, values: Record) -> Self {
89        let updated_fields = values.keys().cloned().collect::<Vec<_>>();
90        let changes = Self::changes_for_fields(None, Some(&values), &updated_fields);
91        Self {
92            kind: RawAuditEventKind::Updated,
93            entity: entity.into(),
94            values: values.clone(),
95            updated_fields,
96            old_values: None,
97            new_values: Some(values),
98            changes,
99            trace_chain: Vec::new(),
100            bootstrap_audit: None,
101        }
102    }
103
104    pub fn updated_with_old_values(
105        entity: impl Into<String>,
106        values: Record,
107        old_values: Option<Record>,
108        new_values: Record,
109        updated_fields: Vec<String>,
110    ) -> Self {
111        let changes =
112            Self::changes_for_fields(old_values.as_ref(), Some(&new_values), &updated_fields);
113        Self {
114            kind: RawAuditEventKind::Updated,
115            entity: entity.into(),
116            values,
117            updated_fields,
118            old_values,
119            new_values: Some(new_values),
120            changes,
121            trace_chain: Vec::new(),
122            bootstrap_audit: None,
123        }
124    }
125
126    pub fn deleted(entity: impl Into<String>, id: Value, expected_version: Option<i64>) -> Self {
127        let mut values = Record::from([("id".to_owned(), id)]);
128        if let Some(version) = expected_version {
129            values.insert("version".to_owned(), Value::I64(version));
130        }
131        Self {
132            kind: RawAuditEventKind::Deleted,
133            entity: entity.into(),
134            values,
135            updated_fields: Vec::new(),
136            old_values: None,
137            new_values: None,
138            changes: Vec::new(),
139            trace_chain: Vec::new(),
140            bootstrap_audit: None,
141        }
142    }
143
144    pub fn deleted_with_old_values(
145        entity: impl Into<String>,
146        id: Value,
147        expected_version: Option<i64>,
148        old_values: Option<Record>,
149    ) -> Self {
150        let mut event = Self::deleted(entity, id, expected_version);
151        event.changes = old_values
152            .as_ref()
153            .map(|values| {
154                values
155                    .iter()
156                    .map(|(field, value)| {
157                        EntityPropertyChange::new(field.clone(), Some(value.clone()), None)
158                    })
159                    .collect()
160            })
161            .unwrap_or_default();
162        event.old_values = old_values;
163        event
164    }
165
166    pub fn recovered(entity: impl Into<String>, id: Value, expected_version: i64) -> Self {
167        let values = Record::from([
168            ("id".to_owned(), id),
169            ("version".to_owned(), Value::I64(expected_version)),
170        ]);
171        Self {
172            kind: RawAuditEventKind::Recovered,
173            entity: entity.into(),
174            values,
175            updated_fields: Vec::new(),
176            old_values: None,
177            new_values: None,
178            changes: Vec::new(),
179            trace_chain: Vec::new(),
180            bootstrap_audit: None,
181        }
182    }
183
184    pub fn recovered_with_old_values(
185        entity: impl Into<String>,
186        id: Value,
187        expected_version: i64,
188        old_values: Option<Record>,
189    ) -> Self {
190        let recovered_version = -expected_version + 1;
191        let mut new_values = old_values.clone().unwrap_or_default();
192        new_values.insert("id".to_owned(), id.clone());
193        new_values.insert("version".to_owned(), Value::I64(recovered_version));
194        let mut event = Self::recovered(entity, id, expected_version);
195        event.old_values = old_values;
196        event.new_values = Some(new_values.clone());
197        event.changes = Self::changes_for_fields(
198            event.old_values.as_ref(),
199            Some(&new_values),
200            &["version".to_owned()],
201        );
202        event
203    }
204
205    /// A new table was created during schema bootstrap.
206    pub fn schema_created(
207        entity: impl Into<String>,
208        table_name: impl Into<String>,
209        field_count: usize,
210    ) -> Self {
211        let entity = entity.into();
212        let values = Record::from([
213            ("table_name".to_owned(), Value::Text(table_name.into())),
214            ("field_count".to_owned(), Value::I64(field_count as i64)),
215        ]);
216        let changes = values
217            .iter()
218            .map(|(k, v)| EntityPropertyChange::new(k.clone(), None, Some(v.clone())))
219            .collect();
220        Self {
221            kind: RawAuditEventKind::SchemaCreated,
222            entity,
223            values,
224            updated_fields: Vec::new(),
225            old_values: None,
226            new_values: None,
227            changes,
228            trace_chain: Vec::new(),
229            bootstrap_audit: None,
230        }
231    }
232
233    /// An existing table was verified during schema bootstrap.
234    pub fn schema_verified(
235        entity: impl Into<String>,
236        table_name: impl Into<String>,
237        field_count: usize,
238    ) -> Self {
239        let entity = entity.into();
240        let values = Record::from([
241            ("table_name".to_owned(), Value::Text(table_name.into())),
242            ("field_count".to_owned(), Value::I64(field_count as i64)),
243        ]);
244        let changes = values
245            .iter()
246            .map(|(k, v)| EntityPropertyChange::new(k.clone(), None, Some(v.clone())))
247            .collect();
248        Self {
249            kind: RawAuditEventKind::SchemaVerified,
250            entity,
251            values,
252            updated_fields: Vec::new(),
253            old_values: None,
254            new_values: None,
255            changes,
256            trace_chain: Vec::new(),
257            bootstrap_audit: None,
258        }
259    }
260
261    /// A new column was added to an existing table (schema migration).
262    pub fn field_added(
263        entity: impl Into<String>,
264        table_name: impl Into<String>,
265        field_name: impl Into<String>,
266    ) -> Self {
267        let entity = entity.into();
268        let values = Record::from([
269            ("table_name".to_owned(), Value::Text(table_name.into())),
270            ("field_name".to_owned(), Value::Text(field_name.into())),
271        ]);
272        let changes = values
273            .iter()
274            .map(|(k, v)| EntityPropertyChange::new(k.clone(), None, Some(v.clone())))
275            .collect();
276        Self {
277            kind: RawAuditEventKind::FieldAdded,
278            entity,
279            values,
280            updated_fields: Vec::new(),
281            old_values: None,
282            new_values: None,
283            changes,
284            trace_chain: Vec::new(),
285            bootstrap_audit: None,
286        }
287    }
288
289    /// Initial seed data was inserted or updated during bootstrap.
290    pub fn data_seeded(
291        entity: impl Into<String>,
292        table_name: impl Into<String>,
293        inserted: usize,
294        updated: usize,
295    ) -> Self {
296        let entity = entity.into();
297        let values = Record::from([
298            ("table_name".to_owned(), Value::Text(table_name.into())),
299            ("inserted".to_owned(), Value::I64(inserted as i64)),
300            ("updated".to_owned(), Value::I64(updated as i64)),
301        ]);
302        let changes = values
303            .iter()
304            .map(|(k, v)| EntityPropertyChange::new(k.clone(), None, Some(v.clone())))
305            .collect();
306        Self {
307            kind: RawAuditEventKind::DataSeeded,
308            entity,
309            values,
310            updated_fields: Vec::new(),
311            old_values: None,
312            new_values: None,
313            changes,
314            trace_chain: Vec::new(),
315            bootstrap_audit: None,
316        }
317    }
318
319    fn changes_for_fields(
320        old_values: Option<&Record>,
321        new_values: Option<&Record>,
322        fields: &[String],
323    ) -> Vec<EntityPropertyChange> {
324        fields
325            .iter()
326            .map(|field| {
327                EntityPropertyChange::new(
328                    field.clone(),
329                    old_values.and_then(|values| values.get(field).cloned()),
330                    new_values.and_then(|values| values.get(field).cloned()),
331                )
332            })
333            .collect()
334    }
335
336    pub fn build_safe_event(
337        &self,
338        audit_mask_fields: &[String],
339        audit_value_max_len: Option<usize>,
340    ) -> SafeAuditEvent {
341        let mut safe_fields = Vec::new();
342        for change in &self.changes {
343            if change.field.starts_with('_') {
344                continue;
345            }
346            // For audit, if it's masked or we just want the new/old values, we should represent it stringified.
347            // Usually we care about the new value in SafeAuditEvent. Or maybe we want to represent the change.
348            // Based on design doc, we stringify the value and apply masks.
349            let raw_val_str = change.new_value.as_ref().map(|v| format!("{:?}", v));
350            let safe_field = build_safe_audit_field(
351                &change.field,
352                raw_val_str.as_deref(),
353                audit_mask_fields,
354                audit_value_max_len,
355            );
356            safe_fields.push(safe_field);
357        }
358
359        SafeAuditEvent {
360            kind: self.kind,
361            entity: self.entity.clone(),
362            fields: safe_fields,
363            trace_chain: self.trace_chain.clone(),
364        }
365    }
366}
367
368pub fn mask_audit_value(value: &str) -> String {
369    let chars: Vec<char> = value.chars().collect();
370    let len = chars.len();
371
372    if len == 0 {
373        return String::new();
374    }
375
376    if chars.iter().all(|c| c.is_ascii_digit()) {
377        return "*".repeat(len);
378    }
379
380    if len < 8 {
381        return "*".repeat(len);
382    }
383
384    let prefix: String = chars[0..2].iter().collect();
385    let suffix: String = chars[len - 2..len].iter().collect();
386    let middle = "*".repeat(len - 4);
387
388    format!("{}{}{}", prefix, middle, suffix)
389}
390
391pub fn limit_audit_value(value: &str, max_len: usize) -> (String, bool) {
392    let chars: Vec<char> = value.chars().collect();
393    let len = chars.len();
394
395    if len <= max_len {
396        return (value.to_string(), false);
397    }
398
399    if max_len <= 3 {
400        return ("*".repeat(max_len), true);
401    }
402
403    let marker = "...";
404    let keep_len = max_len - marker.len();
405    let head_len = keep_len / 2;
406    let tail_len = keep_len - head_len;
407
408    let head: String = chars[0..head_len].iter().collect();
409    let tail: String = chars[len - tail_len..len].iter().collect();
410
411    (format!("{}{}{}", head, marker, tail), true)
412}
413
414pub fn build_safe_audit_field(
415    field_name: &str,
416    raw_value: Option<&str>,
417    audit_mask_fields: &[String],
418    audit_value_max_len: Option<usize>,
419) -> SafeAuditField {
420    match raw_value {
421        None => SafeAuditField {
422            name: field_name.to_string(),
423            value: None,
424            masked: false,
425            truncated: false,
426            raw_length: None,
427            output_length: None,
428            mask_reason: None,
429            truncate_reason: None,
430        },
431        Some(raw) => {
432            let raw_length = raw.chars().count();
433            let should_mask = audit_mask_fields.iter().any(|f| f == field_name);
434
435            let mut value = match should_mask {
436                true => mask_audit_value(raw),
437                false => raw.to_string(),
438            };
439
440            let mut truncated = false;
441            if let Some(max_len) = audit_value_max_len {
442                let result = limit_audit_value(&value, max_len);
443                value = result.0;
444                truncated = result.1;
445            }
446
447            let output_length = value.chars().count();
448
449            SafeAuditField {
450                name: field_name.to_string(),
451                value: Some(value),
452                masked: should_mask,
453                truncated,
454                raw_length: Some(raw_length),
455                output_length: Some(output_length),
456                mask_reason: should_mask.then(|| "_audit_mask_fields".to_string()),
457                truncate_reason: truncated.then(|| "_audit_value_max_len".to_string()),
458            }
459        }
460    }
461}
462
463pub trait RawAuditEventSink: Send + Sync {
464    fn on_event(&self, context: &UserContext, event: &RawAuditEvent) -> Result<(), RuntimeError>;
465}
466
467#[derive(Default, Clone)]
468pub struct InMemoryRawAuditEventSink {
469    sinks: Vec<Arc<dyn RawAuditEventSink>>,
470}
471
472impl InMemoryRawAuditEventSink {
473    pub fn new() -> Self {
474        Self::default()
475    }
476
477    pub fn register(&mut self, sink: impl RawAuditEventSink + 'static) {
478        self.sinks.push(Arc::new(sink));
479    }
480
481    pub fn with_sink(mut self, sink: impl RawAuditEventSink + 'static) -> Self {
482        self.register(sink);
483        self
484    }
485}
486
487impl RawAuditEventSink for InMemoryRawAuditEventSink {
488    fn on_event(&self, context: &UserContext, event: &RawAuditEvent) -> Result<(), RuntimeError> {
489        for sink in &self.sinks {
490            sink.on_event(context, event)?;
491        }
492        Ok(())
493    }
494}
495
496#[derive(Debug, Clone, PartialEq)]
497pub struct SafeAuditField {
498    pub name: String,
499    pub value: Option<String>,
500    pub masked: bool,
501    pub truncated: bool,
502    pub raw_length: Option<usize>,
503    pub output_length: Option<usize>,
504    pub mask_reason: Option<String>,
505    pub truncate_reason: Option<String>,
506}
507
508#[derive(Debug, Clone, PartialEq)]
509pub struct SafeAuditEvent {
510    pub kind: RawAuditEventKind,
511    pub entity: String,
512    pub fields: Vec<SafeAuditField>,
513    pub trace_chain: Vec<teaql_core::TraceNode>,
514}
515
516pub trait SafeAuditEventSink: Send + Sync {
517    fn on_safe_event(
518        &self,
519        context: &crate::UserContext,
520        event: &SafeAuditEvent,
521    ) -> Result<(), crate::RuntimeError>;
522}
teaql_runtime/event.rs

teaql_runtime/
event.rs
