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

teaql_runtime/
event.rs
