1use crate::FaucetError;
23use crate::replication::{BindFormat, BindTarget, format_instant};
24use chrono::{DateTime, Duration, Utc};
25use schemars::JsonSchema;
26use serde::{Deserialize, Serialize};
27
28pub const WINDOW_PLACEHOLDER: &str = "${window}";
32
33pub const WINDOW_START_PLACEHOLDER: &str = "${window.start}";
36
37pub const WINDOW_END_PLACEHOLDER: &str = "${window.end}";
40
41fn default_window_template() -> String {
42 WINDOW_PLACEHOLDER.to_owned()
43}
44
45pub const DEFAULT_MAX_WINDOWS: usize = 10_000;
49
50fn default_max_windows() -> usize {
51 DEFAULT_MAX_WINDOWS
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub struct Window {
57 pub start: DateTime<Utc>,
59 pub end: DateTime<Utc>,
61}
62
63#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
67#[serde(deny_unknown_fields)]
68pub struct WindowBind {
69 #[serde(default)]
72 pub into: BindTarget,
73 #[serde(default)]
76 pub name: String,
77 #[serde(default, skip_serializing_if = "Option::is_none")]
80 pub path: Option<String>,
81 #[serde(default)]
83 pub value_type: crate::replication::BindValueType,
84 #[serde(default = "default_window_template")]
92 pub template: String,
93 #[serde(default)]
95 pub format: BindFormat,
96}
97
98impl Default for WindowBind {
99 fn default() -> Self {
100 Self {
101 into: BindTarget::default(),
102 name: String::new(),
103 path: None,
104 value_type: crate::replication::BindValueType::default(),
105 template: default_window_template(),
106 format: BindFormat::default(),
107 }
108 }
109}
110
111impl WindowBind {
112 pub fn is_unset(&self) -> bool {
114 *self == Self::default()
115 }
116
117 pub fn validate(&self, side: &str) -> Result<(), FaucetError> {
120 crate::replication::validate_bind_placement(
121 &format!("window slicing `{side}`"),
122 self.into,
123 &self.name,
124 self.path.as_deref(),
125 )?;
126 if ![
127 WINDOW_PLACEHOLDER,
128 WINDOW_START_PLACEHOLDER,
129 WINDOW_END_PLACEHOLDER,
130 ]
131 .iter()
132 .any(|p| self.template.contains(p))
133 {
134 return Err(FaucetError::Config(format!(
135 "window slicing: `{side}.template` must contain a `{WINDOW_PLACEHOLDER}`, \
136 `{WINDOW_START_PLACEHOLDER}` or `{WINDOW_END_PLACEHOLDER}` placeholder"
137 )));
138 }
139 Ok(())
140 }
141
142 pub fn render(&self, boundary: DateTime<Utc>) -> String {
144 let formatted = format_instant(boundary, self.format);
145 self.template.replace(WINDOW_PLACEHOLDER, &formatted)
146 }
147
148 pub fn render_window(
151 &self,
152 own: DateTime<Utc>,
153 start: DateTime<Utc>,
154 end: DateTime<Utc>,
155 ) -> String {
156 self.template
157 .replace(
158 WINDOW_START_PLACEHOLDER,
159 &format_instant(start, self.format),
160 )
161 .replace(WINDOW_END_PLACEHOLDER, &format_instant(end, self.format))
162 .replace(WINDOW_PLACEHOLDER, &format_instant(own, self.format))
163 }
164}
165
166#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
168#[serde(deny_unknown_fields)]
169pub struct WindowSpec {
170 pub step: String,
174 pub lower: WindowBind,
176 #[serde(default, skip_serializing_if = "WindowBind::is_unset")]
180 pub upper: WindowBind,
181 #[serde(default, skip_serializing_if = "Option::is_none")]
187 pub granularity: Option<String>,
188 #[serde(default, skip_serializing_if = "Option::is_none")]
191 pub lookback: Option<String>,
192 #[serde(default = "default_max_windows")]
196 pub max_windows: usize,
197}
198
199pub fn parse_step(s: &str) -> Result<Duration, FaucetError> {
203 let s = s.trim();
204 let err = || {
205 FaucetError::Config(format!(
206 "window slicing: '{s}' is not a valid duration — use e.g. 45s, 30m, 6h, 30d"
207 ))
208 };
209 let (num, unit) = match s.chars().last() {
210 Some(c) if c.is_ascii_digit() => (s, "s"),
211 Some(c) => (&s[..s.len() - c.len_utf8()], &s[s.len() - c.len_utf8()..]),
212 None => return Err(err()),
213 };
214 let n: i64 = num.parse().map_err(|_| err())?;
215 if n <= 0 {
216 return Err(FaucetError::Config(format!(
217 "window slicing: duration '{s}' must be positive"
218 )));
219 }
220 let d = match unit {
221 "s" => Duration::try_seconds(n),
222 "m" => Duration::try_minutes(n),
223 "h" => Duration::try_hours(n),
224 "d" => Duration::try_days(n),
225 _ => return Err(err()),
226 };
227 match d {
228 Some(d) if d <= Duration::days(MAX_DURATION_DAYS) => Ok(d),
229 _ => Err(FaucetError::Config(format!(
230 "window slicing: duration '{s}' exceeds the {MAX_DURATION_DAYS}-day maximum"
231 ))),
232 }
233}
234
235const MAX_DURATION_DAYS: i64 = 36_525;
237
238impl WindowSpec {
239 pub fn validate(&self) -> Result<(), FaucetError> {
241 parse_step(&self.step)?;
242 if let Some(g) = &self.granularity
243 && parse_step(g)? > parse_step(&self.step)?
244 {
245 return Err(FaucetError::Config(format!(
246 "window slicing: `granularity` ({g}) must not be longer than `step` ({}), or \
247 every window's upper bound would fall before its start",
248 self.step
249 )));
250 }
251 if let Some(l) = &self.lookback {
252 parse_step(l)?;
253 }
254 self.lower.validate("lower")?;
255 if self.has_upper() {
256 self.upper.validate("upper")?;
257 } else if !self.lower.template.contains(WINDOW_END_PLACEHOLDER) {
258 return Err(FaucetError::Config(format!(
259 "window slicing: `upper` may be omitted only when `lower.template` renders the \
260 window end with `{WINDOW_END_PLACEHOLDER}`; otherwise the window is unbounded above"
261 )));
262 }
263 if self.max_windows == 0 {
264 return Err(FaucetError::Config(
265 "window slicing: `max_windows` must be greater than zero".to_owned(),
266 ));
267 }
268 Ok(())
269 }
270
271 pub fn step_duration(&self) -> Result<Duration, FaucetError> {
273 parse_step(&self.step)
274 }
275
276 pub fn granularity_duration(&self) -> Result<Option<Duration>, FaucetError> {
278 self.granularity.as_deref().map(parse_step).transpose()
279 }
280
281 pub fn lookback_duration(&self) -> Result<Option<Duration>, FaucetError> {
283 self.lookback.as_deref().map(parse_step).transpose()
284 }
285
286 pub fn has_upper(&self) -> bool {
289 !self.upper.is_unset()
290 }
291
292 pub fn rendered_end(&self, w: &Window) -> Result<DateTime<Utc>, FaucetError> {
294 Ok(match self.granularity_duration()? {
295 Some(g) => w.end - g,
296 None => w.end,
297 })
298 }
299
300 pub fn render_lower(&self, w: &Window) -> String {
304 let end = self.rendered_end(w).unwrap_or(w.end);
305 self.lower.render_window(w.start, w.start, end)
306 }
307
308 pub fn render_upper(&self, w: &Window) -> Result<String, FaucetError> {
311 let end = self.rendered_end(w)?;
312 Ok(self.upper.render_window(end, w.start, end))
313 }
314
315 pub fn render_binds(&self, w: &Window) -> Result<Vec<(&WindowBind, String)>, FaucetError> {
318 let end = self.rendered_end(w)?;
319 let mut out = vec![(&self.lower, self.lower.render_window(w.start, w.start, end))];
320 if self.has_upper() {
321 out.push((&self.upper, self.upper.render_window(end, w.start, end)));
322 }
323 Ok(out)
324 }
325}
326
327pub fn enumerate_windows(
335 start: DateTime<Utc>,
336 now: DateTime<Utc>,
337 step: Duration,
338 lookback: Option<Duration>,
339 max_windows: usize,
340) -> (Vec<Window>, bool) {
341 let mut cur = match lookback {
342 Some(lb) => start
343 .checked_sub_signed(lb)
344 .unwrap_or(DateTime::<Utc>::MIN_UTC),
345 None => start,
346 };
347 let mut out = Vec::new();
348 let mut truncated = false;
349 while cur < now {
350 if out.len() >= max_windows {
351 truncated = true;
352 break;
353 }
354 let end = cur.checked_add_signed(step).map_or(now, |e| e.min(now));
355 if end <= cur {
359 break;
360 }
361 out.push(Window { start: cur, end });
362 cur = end;
363 }
364 (out, truncated)
365}
366
367#[cfg(test)]
368mod tests {
369 use super::*;
370 use chrono::TimeZone;
371 use serde_json::json;
372
373 fn dt(s: &str) -> DateTime<Utc> {
374 DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)
375 }
376
377 #[test]
378 fn parse_step_units() {
379 assert_eq!(parse_step("45s").unwrap(), Duration::seconds(45));
380 assert_eq!(parse_step("30m").unwrap(), Duration::minutes(30));
381 assert_eq!(parse_step("6h").unwrap(), Duration::hours(6));
382 assert_eq!(parse_step("30d").unwrap(), Duration::days(30));
383 assert_eq!(parse_step("3600").unwrap(), Duration::seconds(3600));
384 }
385
386 #[test]
387 fn parse_step_rejects_bad() {
388 assert!(parse_step("0d").is_err());
389 assert!(parse_step("-1h").is_err());
390 assert!(parse_step("").is_err());
391 assert!(parse_step("10y").is_err());
392 assert!(parse_step("abc").is_err());
393 }
394
395 #[test]
396 fn enumerate_contiguous_half_open() {
397 let (ws, trunc) = enumerate_windows(
398 dt("2024-01-01T00:00:00Z"),
399 dt("2024-01-04T00:00:00Z"),
400 Duration::days(1),
401 None,
402 100,
403 );
404 assert!(!trunc);
405 assert_eq!(ws.len(), 3);
406 assert_eq!(ws[0].start, dt("2024-01-01T00:00:00Z"));
407 assert_eq!(ws[0].end, dt("2024-01-02T00:00:00Z"));
408 assert_eq!(ws[0].end, ws[1].start);
410 assert_eq!(ws[2].end, dt("2024-01-04T00:00:00Z"));
411 }
412
413 #[test]
414 fn last_window_clamps_to_now() {
415 let (ws, _) = enumerate_windows(
416 dt("2024-01-01T00:00:00Z"),
417 dt("2024-01-02T06:00:00Z"),
418 Duration::days(1),
419 None,
420 100,
421 );
422 assert_eq!(ws.len(), 2);
423 assert_eq!(ws[1].start, dt("2024-01-02T00:00:00Z"));
424 assert_eq!(ws[1].end, dt("2024-01-02T06:00:00Z")); }
426
427 #[test]
428 fn empty_when_start_at_or_after_now() {
429 let (ws, trunc) = enumerate_windows(
430 dt("2024-06-01T00:00:00Z"),
431 dt("2024-06-01T00:00:00Z"),
432 Duration::days(1),
433 None,
434 100,
435 );
436 assert!(ws.is_empty());
437 assert!(!trunc);
438 }
439
440 #[test]
441 fn lookback_extends_the_first_window_backwards() {
442 let (ws, _) = enumerate_windows(
443 dt("2024-01-02T00:00:00Z"),
444 dt("2024-01-03T00:00:00Z"),
445 Duration::days(1),
446 Some(Duration::hours(6)),
447 100,
448 );
449 assert_eq!(ws[0].start, dt("2024-01-01T18:00:00Z"));
451 }
452
453 #[test]
454 fn max_windows_truncates_and_flags() {
455 let (ws, trunc) = enumerate_windows(
456 dt("2024-01-01T00:00:00Z"),
457 dt("2024-12-31T00:00:00Z"),
458 Duration::days(1),
459 None,
460 5,
461 );
462 assert_eq!(ws.len(), 5);
463 assert!(trunc);
464 assert_eq!(ws[4].end, dt("2024-01-06T00:00:00Z"));
466 }
467
468 #[test]
469 fn render_lower_and_upper_with_granularity() {
470 let spec = WindowSpec {
471 step: "1d".into(),
472 lower: WindowBind {
473 into: BindTarget::Query,
474 name: "start".into(),
475 template: "${window}".into(),
476 format: BindFormat::Date,
477 path: None,
478 value_type: Default::default(),
479 },
480 upper: WindowBind {
481 into: BindTarget::Query,
482 name: "end".into(),
483 template: "${window}".into(),
484 format: BindFormat::Date,
485 path: None,
486 value_type: Default::default(),
487 },
488 granularity: Some("1d".into()),
489 lookback: None,
490 max_windows: DEFAULT_MAX_WINDOWS,
491 };
492 let w = Window {
493 start: dt("2024-01-01T00:00:00Z"),
494 end: dt("2024-01-02T00:00:00Z"),
495 };
496 assert_eq!(spec.render_lower(&w), "2024-01-01");
497 assert_eq!(spec.render_upper(&w).unwrap(), "2024-01-01");
499 }
500
501 #[test]
502 fn render_template_and_epoch_format() {
503 let bind = WindowBind {
504 into: BindTarget::Query,
505 name: "since".into(),
506 template: "gte|${window}".into(),
507 format: BindFormat::EpochS,
508 path: None,
509 value_type: Default::default(),
510 };
511 let ts = Utc.timestamp_opt(1_700_000_000, 0).unwrap();
512 assert_eq!(bind.render(ts), "gte|1700000000");
513 }
514
515 #[test]
516 fn validate_catches_misconfig() {
517 let ok = WindowSpec {
518 step: "1d".into(),
519 lower: WindowBind {
520 into: BindTarget::Query,
521 name: "start".into(),
522 template: "${window}".into(),
523 format: BindFormat::Iso8601,
524 path: None,
525 value_type: Default::default(),
526 },
527 upper: WindowBind {
528 into: BindTarget::Query,
529 name: "end".into(),
530 template: "${window}".into(),
531 format: BindFormat::Iso8601,
532 path: None,
533 value_type: Default::default(),
534 },
535 granularity: None,
536 lookback: None,
537 max_windows: DEFAULT_MAX_WINDOWS,
538 };
539 ok.validate().unwrap();
540
541 let mut bad_step = ok.clone();
542 bad_step.step = "0d".into();
543 assert!(bad_step.validate().is_err());
544
545 let mut empty_name = ok.clone();
546 empty_name.lower.name = " ".into();
547 assert!(empty_name.validate().is_err());
548
549 let mut no_placeholder = ok.clone();
550 no_placeholder.upper.template = "fixed".into();
551 assert!(no_placeholder.validate().is_err());
552
553 let mut zero_windows = ok.clone();
554 zero_windows.max_windows = 0;
555 assert!(zero_windows.validate().is_err());
556 }
557
558 #[test]
559 fn spec_deserializes_from_yaml_shape() {
560 let v = json!({
561 "step": "30d",
562 "lower": {"into": "query", "name": "start_date", "format": "date"},
563 "upper": {"into": "query", "name": "end_date", "format": "date"},
564 "lookback": "1d"
565 });
566 let spec: WindowSpec = serde_json::from_value(v).unwrap();
567 assert_eq!(spec.step, "30d");
568 assert_eq!(spec.lower.template, WINDOW_PLACEHOLDER); assert_eq!(spec.max_windows, DEFAULT_MAX_WINDOWS); spec.validate().unwrap();
571 }
572
573 #[test]
574 fn absurd_durations_are_config_errors_not_panics() {
575 assert!(parse_step("9223372036854775807d").is_err());
576 assert!(parse_step("40000d").is_err());
577 assert!(parse_step("36525d").is_ok());
578 let now = Utc::now();
579 let (ws, _) = enumerate_windows(now, now, Duration::days(1), Some(Duration::MAX), 3);
580 assert_eq!(ws.len(), 3);
581 assert_eq!(ws[0].start, DateTime::<Utc>::MIN_UTC);
582 let start = now - Duration::days(2);
583 let (ws, _) = enumerate_windows(start, now, Duration::MAX, None, 10);
584 assert_eq!(ws, vec![Window { start, end: now }]);
585 }
586
587 #[test]
588 fn granularity_must_not_exceed_step() {
589 let mut spec = combined_spec();
590 spec.validate().unwrap();
591 spec.granularity = Some("7d".into());
592 spec.validate()
593 .expect("one granule per window: an inclusive upper bound equal to the start");
594 spec.granularity = Some("8d".into());
595 let err = spec.validate().unwrap_err();
596 assert!(err.to_string().contains("longer than `step`"), "{err}");
597 }
598
599 fn combined_spec() -> WindowSpec {
600 serde_json::from_value(json!({
601 "step": "7d",
602 "granularity": "1d",
603 "lower": {
604 "into": "body", "path": "/query", "format": "date",
605 "template": "WHERE d BETWEEN '${window.start}' AND '${window.end}'"
606 }
607 }))
608 .unwrap()
609 }
610
611 #[test]
612 fn combined_bind_renders_both_bounds_with_granularity() {
613 let spec = combined_spec();
614 spec.validate().unwrap();
615 assert!(!spec.has_upper());
616 let w = Window {
617 start: dt("2026-09-01T00:00:00Z"),
618 end: dt("2026-09-08T00:00:00Z"),
619 };
620 let want = "WHERE d BETWEEN '2026-09-01' AND '2026-09-07'";
621 assert_eq!(spec.render_lower(&w), want);
622 let binds = spec.render_binds(&w).unwrap();
623 assert_eq!(binds.len(), 1);
624 assert_eq!(binds[0].1, want);
625 let back = serde_json::to_value(&spec).unwrap();
626 assert!(back.get("upper").is_none());
627 }
628
629 #[test]
630 fn omitted_upper_requires_window_end_in_lower() {
631 let mut spec = combined_spec();
632 spec.lower.template = "d >= '${window.start}'".into();
633 let err = spec.validate().unwrap_err().to_string();
634 assert!(err.contains("unbounded above"), "{err}");
635 spec.lower.template = "d >= '${window}'".into();
636 assert!(spec.validate().is_err());
637 }
638
639 #[test]
640 fn two_bind_spec_renders_both_and_accepts_named_placeholders() {
641 let mut spec: WindowSpec = serde_json::from_value(json!({
642 "step": "1d",
643 "granularity": "1s",
644 "lower": {"into": "query", "name": "from"},
645 "upper": {"into": "query", "name": "to", "template": "${window.start}..${window.end}"}
646 }))
647 .unwrap();
648 spec.validate().unwrap();
649 let w = Window {
650 start: dt("2026-01-01T00:00:00Z"),
651 end: dt("2026-01-02T00:00:00Z"),
652 };
653 let binds = spec.render_binds(&w).unwrap();
654 assert_eq!(binds.len(), 2);
655 assert_eq!(binds[0].1, spec.render_lower(&w));
656 assert_eq!(binds[1].1, spec.render_upper(&w).unwrap());
657 assert!(binds[1].1.contains(".."));
658 spec.granularity = Some("bad".into());
659 assert!(spec.render_binds(&w).is_err());
660 assert_eq!(spec.render_lower(&w), spec.lower.render(w.start));
661 assert!(WindowBind::default().is_unset());
662 }
663}