1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
//! Soft delete handling for the pipeline
use crate::config::SoftDeleteConfig;
use crate::models::{stream_event::Event, EventType};
use serde_json::Value;
use tracing::{debug, trace};
/// Handles soft delete transformations
#[derive(Clone)]
pub struct SoftDeleteHandler {
config: SoftDeleteConfig,
}
impl SoftDeleteHandler {
pub fn new(config: SoftDeleteConfig) -> Self {
Self { config }
}
/// Check if an event should be transformed to a delete based on soft delete rules
pub fn transform_event(&self, event: Event) -> Option<Event> {
match event {
Event::Cdc(ref cdc_event) => {
// Only process UPDATE events for soft delete
if cdc_event.event_type != EventType::Update {
return Some(event);
}
// Check if CDC handling is enabled
if !self.config.handle_on_cdc {
return Some(event);
}
// Check if the field value matches any delete value
if let Some(field_value) = cdc_event.data.get(&self.config.field) {
if self.is_soft_deleted(field_value) {
debug!(
"Transforming UPDATE to DELETE for table '{}' due to soft delete field '{}' = {:?}",
cdc_event.table, self.config.field, field_value
);
// Transform to a Delete event
return Some(Event::Delete {
table: cdc_event.table.clone(),
old_data: Value::Object(cdc_event.data.clone().into_iter().collect()),
position: cdc_event.position.clone(),
});
}
}
Some(event)
}
Event::Update {
ref table,
ref new_data,
ref old_data,
ref position,
} => {
// Check if CDC handling is enabled
if !self.config.handle_on_cdc {
return Some(event);
}
// Check if the field value matches any delete value
if let Some(field_value) = new_data.get(&self.config.field) {
if self.is_soft_deleted(field_value) {
debug!(
"Transforming UPDATE to DELETE for table '{}' due to soft delete field '{}' = {:?}",
table, self.config.field, field_value
);
// Transform to a Delete event
return Some(Event::Delete {
table: table.clone(),
old_data: old_data.clone(),
position: position.clone(),
});
}
}
Some(event)
}
Event::FullSync {
ref table,
ref data,
} => {
// Check if full sync handling is enabled
if !self.config.handle_on_full_sync {
return Some(event);
}
// Check if the field value matches any delete value
if let Some(field_value) = data.get(&self.config.field) {
if self.is_soft_deleted(field_value) {
trace!(
"Filtering out soft-deleted record from full sync for table '{}': {} = {:?}",
table, self.config.field, field_value
);
// Filter out soft-deleted records during full sync
return None;
}
}
Some(event)
}
// Pass through other events unchanged
_ => Some(event),
}
}
/// Check if a value matches any of the configured delete values
fn is_soft_deleted(&self, value: &Value) -> bool {
self.config.delete_values.iter().any(|delete_value| {
// Use serde_json's equality comparison which handles different types
value == delete_value
})
}
}