oxirs_stream/processing/
window.rs

1//! Window management for event processing
2//!
3//! This module provides windowing capabilities for stream processing including:
4//! - Time-based windowing (tumbling, sliding)
5//! - Count-based windowing
6//! - Session-based windowing
7//! - Custom window types
8
9use crate::StreamEvent;
10use anyhow::Result;
11use chrono::{DateTime, Duration as ChronoDuration, Utc};
12use serde::{Deserialize, Serialize};
13use std::collections::{HashMap, VecDeque};
14use uuid::Uuid;
15
16/// Window types for event processing
17#[derive(Debug, Clone, Serialize, Deserialize)]
18pub enum WindowType {
19    /// Fixed time-based window
20    Tumbling { duration: ChronoDuration },
21    /// Overlapping time-based window
22    Sliding {
23        duration: ChronoDuration,
24        slide: ChronoDuration,
25    },
26    /// Count-based window
27    CountBased { size: usize },
28    /// Session-based window (events grouped by activity)
29    Session { timeout: ChronoDuration },
30    /// Custom window with user-defined logic
31    Custom { name: String },
32}
33
34/// Window trigger conditions
35#[derive(Debug, Clone, Serialize, Deserialize)]
36pub enum WindowTrigger {
37    /// Trigger when window ends
38    OnTime,
39    /// Trigger every N events
40    OnCount(usize),
41    /// Trigger on specific conditions
42    OnCondition(String),
43    /// Trigger both on time and count
44    Hybrid { time: ChronoDuration, count: usize },
45}
46
47/// Window configuration
48#[derive(Debug, Clone, Serialize, Deserialize)]
49pub struct WindowConfig {
50    pub window_type: WindowType,
51    pub aggregates: Vec<super::aggregation::AggregateFunction>,
52    pub group_by: Vec<String>,
53    pub filter: Option<String>,
54    pub allow_lateness: Option<ChronoDuration>,
55    pub trigger: WindowTrigger,
56}
57
58/// Event processing window
59#[derive(Debug)]
60pub struct EventWindow {
61    id: String,
62    config: WindowConfig,
63    events: VecDeque<StreamEvent>,
64    start_time: DateTime<Utc>,
65    end_time: Option<DateTime<Utc>>,
66    last_trigger: Option<DateTime<Utc>>,
67    event_count: usize,
68    aggregation_state: HashMap<String, super::aggregation::AggregationState>,
69}
70
71/// Result of window aggregation
72#[derive(Debug, Clone, Serialize, Deserialize)]
73pub struct WindowResult {
74    pub window_id: String,
75    pub window_start: DateTime<Utc>,
76    pub window_end: DateTime<Utc>,
77    pub event_count: usize,
78    pub aggregations: HashMap<String, serde_json::Value>,
79    pub trigger_reason: String,
80    pub processing_time: DateTime<Utc>,
81}
82
83impl EventWindow {
84    /// Create a new event window
85    pub fn new(config: WindowConfig) -> Self {
86        let id = Uuid::new_v4().to_string();
87        let start_time = Utc::now();
88
89        Self {
90            id,
91            config,
92            events: VecDeque::new(),
93            start_time,
94            end_time: None,
95            last_trigger: None,
96            event_count: 0,
97            aggregation_state: HashMap::new(),
98        }
99    }
100
101    /// Add an event to the window
102    pub fn add_event(&mut self, event: StreamEvent) -> Result<()> {
103        self.events.push_back(event);
104        self.event_count += 1;
105
106        // Update aggregation state
107        self.update_aggregations()?;
108
109        Ok(())
110    }
111
112    /// Check if window should trigger
113    pub fn should_trigger(&self, current_time: DateTime<Utc>) -> bool {
114        match &self.config.trigger {
115            WindowTrigger::OnTime => match &self.config.window_type {
116                WindowType::Tumbling { duration } => current_time >= self.start_time + *duration,
117                WindowType::Sliding { duration, .. } => current_time >= self.start_time + *duration,
118                _ => false,
119            },
120            WindowTrigger::OnCount(count) => self.event_count >= *count,
121            WindowTrigger::OnCondition(condition) => self.evaluate_condition(condition),
122            WindowTrigger::Hybrid { time, count } => {
123                let time_condition = current_time >= self.start_time + *time;
124                let count_condition = self.event_count >= *count;
125                time_condition || count_condition
126            }
127        }
128    }
129
130    /// Evaluate trigger condition
131    fn evaluate_condition(&self, condition: &str) -> bool {
132        match condition {
133            "window_full" => match &self.config.window_type {
134                WindowType::CountBased { size } => self.event_count >= *size,
135                _ => false,
136            },
137            "always" => true,
138            "never" => false,
139            condition if condition.starts_with("time_elapsed:") => {
140                if let Ok(seconds) = condition
141                    .strip_prefix("time_elapsed:")
142                    .expect("strip_prefix should succeed after starts_with check")
143                    .parse::<i64>()
144                {
145                    let duration = ChronoDuration::seconds(seconds);
146                    Utc::now() >= self.start_time + duration
147                } else {
148                    false
149                }
150            }
151            condition if condition.starts_with("count_gte:") => {
152                if let Ok(count) = condition
153                    .strip_prefix("count_gte:")
154                    .expect("strip_prefix should succeed after starts_with check")
155                    .parse::<usize>()
156                {
157                    self.event_count >= count
158                } else {
159                    false
160                }
161            }
162            condition if condition.starts_with("count_eq:") => {
163                if let Ok(count) = condition
164                    .strip_prefix("count_eq:")
165                    .expect("strip_prefix should succeed after starts_with check")
166                    .parse::<usize>()
167                {
168                    self.event_count == count
169                } else {
170                    false
171                }
172            }
173            _ => condition.parse::<bool>().unwrap_or_default(),
174        }
175    }
176
177    /// Update aggregation state with the most recently added event.
178    ///
179    /// For every aggregate function configured on this window we lazily
180    /// initialize an [`AggregationState`] (keyed by a stable function name) and
181    /// fold the just-added event into it. This is what makes windowed
182    /// Count/Sum/Average/Min/Max/Distinct actually compute instead of returning
183    /// an empty map.
184    fn update_aggregations(&mut self) -> Result<()> {
185        // The event we just added is at the back of the queue.
186        let event = match self.events.back() {
187            Some(event) => event.clone(),
188            None => return Ok(()),
189        };
190
191        for function in &self.config.aggregates {
192            let key = aggregate_state_key(function);
193            let state = self
194                .aggregation_state
195                .entry(key)
196                .or_insert_with(|| super::aggregation::AggregationState::new(function));
197            state.update(&event, function)?;
198        }
199
200        Ok(())
201    }
202
203    /// Compute the current aggregation results for this window.
204    ///
205    /// Each configured aggregate is emitted under the same stable key used by
206    /// `update_aggregations`.
207    pub fn aggregation_results(&self) -> Result<HashMap<String, serde_json::Value>> {
208        let mut results = HashMap::new();
209        for (name, state) in &self.aggregation_state {
210            results.insert(name.clone(), state.result()?);
211        }
212        Ok(results)
213    }
214
215    /// Get window ID
216    pub fn id(&self) -> &str {
217        &self.id
218    }
219
220    /// Get window configuration
221    pub fn config(&self) -> &WindowConfig {
222        &self.config
223    }
224
225    /// Get events in window
226    pub fn events(&self) -> &VecDeque<StreamEvent> {
227        &self.events
228    }
229
230    /// Get event count
231    pub fn event_count(&self) -> usize {
232        self.event_count
233    }
234
235    /// Get aggregation state
236    pub fn aggregation_state(&self) -> &HashMap<String, super::aggregation::AggregationState> {
237        &self.aggregation_state
238    }
239}
240
241/// Derive a stable state key for an aggregate function so that repeated updates
242/// and result reads address the same [`AggregationState`] entry.
243fn aggregate_state_key(function: &super::aggregation::AggregateFunction) -> String {
244    use super::aggregation::AggregateFunction as Af;
245    match function {
246        Af::Count => "count".to_string(),
247        Af::Sum { field } => format!("sum:{field}"),
248        Af::Average { field } => format!("avg:{field}"),
249        Af::Min { field } => format!("min:{field}"),
250        Af::Max { field } => format!("max:{field}"),
251        Af::First => "first".to_string(),
252        Af::Last => "last".to_string(),
253        Af::Distinct { field } => format!("distinct:{field}"),
254        Af::Custom { name, .. } => format!("custom:{name}"),
255    }
256}
257
258/// Watermark for tracking event time progress
259#[derive(Debug, Clone, Serialize, Deserialize)]
260pub struct Watermark {
261    /// Current watermark timestamp
262    pub timestamp: DateTime<Utc>,
263    /// Allowed lateness after watermark
264    pub allowed_lateness: ChronoDuration,
265}
266
267impl Watermark {
268    /// Create a new watermark with default values
269    pub fn new() -> Self {
270        Self {
271            timestamp: Utc::now(),
272            allowed_lateness: ChronoDuration::seconds(60),
273        }
274    }
275
276    /// Update watermark with new timestamp
277    pub fn update(&mut self, timestamp: DateTime<Utc>) {
278        if timestamp > self.timestamp {
279            self.timestamp = timestamp;
280        }
281    }
282
283    /// Get the current watermark timestamp
284    pub fn current(&self) -> DateTime<Utc> {
285        self.timestamp
286    }
287}
288
289impl Default for Watermark {
290    fn default() -> Self {
291        Self::new()
292    }
293}
294
295#[cfg(test)]
296mod tests {
297    use super::super::aggregation::AggregateFunction;
298    use super::*;
299    use crate::event::EventMetadata;
300
301    fn triple(subject: &str) -> StreamEvent {
302        StreamEvent::TripleAdded {
303            subject: subject.to_string(),
304            predicate: "http://example.org/p".to_string(),
305            object: "http://example.org/o".to_string(),
306            graph: None,
307            metadata: EventMetadata::default(),
308        }
309    }
310
311    #[test]
312    fn regression_window_aggregations_actually_compute() {
313        let config = WindowConfig {
314            window_type: WindowType::CountBased { size: 10 },
315            aggregates: vec![
316                AggregateFunction::Count,
317                AggregateFunction::Distinct {
318                    field: "subject".to_string(),
319                },
320            ],
321            group_by: vec![],
322            filter: None,
323            allow_lateness: None,
324            trigger: WindowTrigger::OnCount(10),
325        };
326
327        let mut window = EventWindow::new(config);
328        window.add_event(triple("http://example.org/s1")).unwrap();
329        window.add_event(triple("http://example.org/s1")).unwrap();
330        window.add_event(triple("http://example.org/s2")).unwrap();
331
332        let results = window.aggregation_results().unwrap();
333
334        // Count must reflect all three events (previously always empty).
335        assert_eq!(
336            results.get("count"),
337            Some(&serde_json::Value::Number(3u64.into()))
338        );
339        // Distinct subjects: s1, s2 => 2 distinct values.
340        assert_eq!(
341            results.get("distinct:subject"),
342            Some(&serde_json::Value::Number(2u64.into()))
343        );
344    }
345}
oxirs_stream/processing/window.rs

oxirs_stream/processing/
window.rs
