reifydb-flow 0.9.1

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use reifydb_value::value::{datetime::DateTime, duration::Duration};

use crate::operator::state::seal::coord::Coord;

pub const SEAL_GATE_STEP: Duration = Duration::from_milliseconds_const(1);

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct AdmissibleSpan(Duration);

impl AdmissibleSpan {
	pub fn duration(self) -> Duration {
		self.0
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct SealInstant(DateTime);

impl SealInstant {
	pub fn at(self) -> DateTime {
		self.0
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct EvictionInstant(DateTime);

impl EvictionInstant {
	pub fn at(self) -> DateTime {
		self.0
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct SealedThrough(DateTime);

impl SealedThrough {
	pub fn from_order(order: u64) -> Self {
		Self(<DateTime as Coord>::from_order(order))
	}

	pub fn at(self) -> DateTime {
		self.0
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SealRule {
	admissible: AdmissibleSpan,
}

impl SealRule {
	pub fn tumbling(size: Duration, lateness: Duration) -> Self {
		Self::extended_by_seal(size, lateness)
	}

	pub fn sliding(size: Duration, lateness: Duration) -> Self {
		Self::extended_by_seal(size, lateness)
	}

	pub fn session(gap: Duration, lateness: Duration) -> Self {
		Self::extended_by_seal(gap, lateness)
	}

	pub fn rolling(span: Duration, lateness: Duration) -> Self {
		Self::extended_by_seal(span, lateness)
	}

	pub fn of(admissible: Duration) -> Self {
		Self {
			admissible: AdmissibleSpan(admissible),
		}
	}

	fn extended_by_seal(base: Duration, lateness: Duration) -> Self {
		Self {
			admissible: AdmissibleSpan(base.try_add(lateness).unwrap_or(base)),
		}
	}

	pub fn admissible(self) -> AdmissibleSpan {
		self.admissible
	}

	pub fn is_inert(self) -> bool {
		self.admissible.0.is_zero()
	}

	pub fn seal_instant(self, event: DateTime) -> SealInstant {
		SealInstant(event.saturating_add(self.admissible.0).saturating_add(SEAL_GATE_STEP))
	}

	pub fn seal_instant_from_order(self, event_order: u64) -> SealInstant {
		self.seal_instant(<DateTime as Coord>::from_order(event_order))
	}

	pub fn horizon_at(self, at: DateTime) -> Option<DateTime> {
		at.checked_sub(self.admissible.0).and_then(|horizon| horizon.checked_sub(SEAL_GATE_STEP))
	}
}

pub fn seal_horizon<C: Coord>(watermark: C, lateness: C::Span) -> C {
	watermark.saturating_sub_span(lateness)
}

pub fn is_sealed<C: Coord>(event: C, horizon: C) -> bool {
	event < horizon
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EvictionRule {
	span: Duration,
}

impl EvictionRule {
	pub fn rolling(span: Duration) -> Self {
		Self {
			span,
		}
	}

	pub fn eviction_instant(self, event: DateTime) -> EvictionInstant {
		EvictionInstant(event.saturating_add(self.span))
	}

	pub fn eviction_instant_from_order(self, event_order: u64) -> EvictionInstant {
		self.eviction_instant(<DateTime as Coord>::from_order(event_order))
	}
}

#[cfg(test)]
mod tests {
	use reifydb_value::factory::time::at_millis;

	use super::*;

	fn ms(millis: u64) -> Duration {
		Duration::from_milliseconds_const(millis as i64)
	}

	#[test]
	fn a_seal_instant_is_one_past_the_admissible_span() {
		// The wheel fires inclusively (`at <= watermark`) but the seal gate is strict - a window
		// closes once the watermark has passed its whole admissible span, not on reaching it. The +1
		// is what converts one into the other.
		let rule = SealRule::tumbling(ms(1_000), ms(200));

		assert_eq!(rule.admissible().duration(), ms(1_200));
		assert_eq!(rule.seal_instant(at_millis(5_000)).at(), at_millis(6_201));
	}

	#[test]
	fn the_horizon_trails_the_ledger_by_the_whole_admissible_span() {
		// The ledger holds the instant a seal timer fired, a whole admissible span ahead of the
		// newest window that timer actually sealed. Treating the ledger itself as the immutable
		// frontier erases the accumulator of a window that is still open and still taking rows.
		let rule = SealRule::tumbling(ms(30_000), ms(45_000));
		let ledger = at_millis(358_262);

		let horizon = rule.horizon_at(ledger).expect("the ledger is past one admissible span");

		assert_eq!(horizon, at_millis(283_261), "ledger - (size + lateness) - 1");
		assert!(horizon < ledger, "a frontier at or past the ledger reclaims windows that have not sealed");
		assert_eq!(
			rule.seal_instant(horizon).at(),
			ledger,
			"and it is the exact inverse of arming, so a window at this horizon sealed at precisely \
			 this ledger rather than one millisecond either side of it"
		);
	}

	#[test]
	fn a_ledger_short_of_one_admissible_span_has_sealed_nothing() {
		// Early in a operator's life the ledger sits below its own span. Wrapping through u64 would put
		// the horizon near u64::MAX and report every window sealed, reclaiming the operator in one sweep.
		let rule = SealRule::tumbling(ms(30_000), ms(45_000));

		assert_eq!(rule.horizon_at(at_millis(0)), None);
		assert_eq!(rule.horizon_at(at_millis(75_000)), None, "the horizon would be 0 - 1, not 0");
		assert_eq!(rule.horizon_at(at_millis(75_001)), Some(at_millis(0)));
	}

	#[test]
	fn rolling_admission_carries_lateness_and_rolling_eviction_does_not() {
		// Rolling admits a late event inside the lateness but evicts on the bare span, which is why
		// SealInstant and EvictionInstant are separate types. An eviction that also waited out the
		// lateness keeps every rolling window one lateness-period too wide, inflating every aggregate.
		let admission = SealRule::rolling(ms(1_000), ms(200));
		let eviction = EvictionRule::rolling(ms(1_000));

		assert_eq!(admission.seal_instant(at_millis(5_000)).at(), at_millis(6_201));
		assert_eq!(eviction.eviction_instant(at_millis(5_000)).at(), at_millis(6_000));
	}

	#[test]
	fn an_eviction_instant_never_carries_the_strict_gate_plus_one() {
		// The +1 belongs to the seal gate alone. Eviction is a retention boundary, not a gate, so
		// carrying the +1 there retains one millisecond too much on every rolling window, forever.
		let eviction = EvictionRule::rolling(ms(0));

		assert_eq!(eviction.eviction_instant(at_millis(7_000)).at(), at_millis(7_000));
	}

	#[test]
	fn every_kind_admits_its_own_base_span_plus_lateness() {
		// Tumbling and sliding admit size + lateness, session admits gap + lateness, rolling
		// admits span + lateness. A divergence here is a behaviour change, not a refactor.
		assert_eq!(SealRule::tumbling(ms(1_000), ms(50)).admissible().duration(), ms(1_050));
		assert_eq!(SealRule::sliding(ms(1_000), ms(50)).admissible().duration(), ms(1_050));
		assert_eq!(SealRule::session(ms(300), ms(50)).admissible().duration(), ms(350));
		assert_eq!(SealRule::rolling(ms(2_000), ms(50)).admissible().duration(), ms(2_050));
	}

	#[test]
	fn no_lateness_can_make_the_admissible_span_shorter_than_the_window() {
		// An admissible span below the window size seals live windows on arrival - silent data loss.
		// Two ways to break it: the sum failing back to something smaller than the base, and
		// span_millis answering none for a months/days Duration, which i64::MAX nanoseconds becomes.
		let enormous = Duration::from_nanoseconds_const(i64::MAX);

		for lateness in [ms(0), ms(1), enormous] {
			let rule = SealRule::tumbling(ms(1_000), lateness);
			assert!(
				rule.admissible().duration() >= ms(1_000),
				"admissible {:?} fell below the 1000ms window for lateness {lateness:?}",
				rule.admissible().duration()
			);
		}
	}
}