reifydb-flow 0.9.1

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

use crate::window::{
	engine::KeyspaceFamily,
	span::{SlotSpan, WindowAnchor},
};

pub const DEFAULT_EXPIRE_BATCH: usize = 256;

#[derive(Clone)]
pub struct WindowEngineConfig {
	expire_batch: usize,
	family: KeyspaceFamily,
}

impl WindowEngineConfig {
	pub fn builder() -> WindowEngineConfigBuilder {
		WindowEngineConfigBuilder::new()
	}

	pub fn expire_batch(&self) -> usize {
		self.expire_batch
	}

	pub fn family(&self) -> KeyspaceFamily {
		self.family
	}
}

pub struct WindowEngineConfigBuilder {
	expire_batch: usize,
	family: KeyspaceFamily,
}

impl WindowEngineConfigBuilder {
	fn new() -> Self {
		Self {
			expire_batch: DEFAULT_EXPIRE_BATCH,
			family: KeyspaceFamily::Host,
		}
	}

	pub fn expire_batch(mut self, batch: usize) -> Self {
		self.expire_batch = batch;
		self
	}

	pub fn family(mut self, family: KeyspaceFamily) -> Self {
		self.family = family;
		self
	}

	pub fn build(self) -> WindowEngineConfig {
		WindowEngineConfig {
			expire_batch: self.expire_batch,
			family: self.family,
		}
	}
}

pub struct TumblingCarryConfig<S: WindowAnchor> {
	base: WindowEngineConfig,
	retention: Option<SlotSpan<S>>,
}

impl<S: WindowAnchor> TumblingCarryConfig<S> {
	pub fn builder(base: WindowEngineConfig) -> TumblingCarryConfigBuilder<S> {
		TumblingCarryConfigBuilder::new(base)
	}

	pub fn base(&self) -> WindowEngineConfig {
		self.base.clone()
	}

	pub fn retention(&self) -> Option<SlotSpan<S>> {
		self.retention
	}
}

pub struct TumblingCarryConfigBuilder<S: WindowAnchor> {
	base: WindowEngineConfig,
	retention: Option<SlotSpan<S>>,
}

impl<S: WindowAnchor> TumblingCarryConfigBuilder<S> {
	fn new(base: WindowEngineConfig) -> Self {
		Self {
			base,
			retention: None,
		}
	}

	pub fn retention(mut self, retention: Option<SlotSpan<S>>) -> Self {
		self.retention = retention;
		self
	}

	pub fn build(self) -> TumblingCarryConfig<S> {
		TumblingCarryConfig {
			base: self.base,
			retention: self.retention,
		}
	}
}

#[cfg(test)]
mod tests {
	use reifydb_value::value::datetime::DateTime;

	use super::*;

	#[test]
	fn the_expire_batch_defaults_and_survives_an_override() {
		// The batch bounds one expiry pass; a builder that dropped it would sweep unbounded.
		assert_eq!(WindowEngineConfig::builder().build().expire_batch(), DEFAULT_EXPIRE_BATCH);
		assert_eq!(WindowEngineConfig::builder().expire_batch(9).build().expire_batch(), 9);
	}

	#[test]
	fn a_carry_config_forwards_its_base() {
		let config: TumblingCarryConfig<DateTime> =
			TumblingCarryConfig::builder(WindowEngineConfig::builder().expire_batch(7).build())
				.retention(None)
				.build();

		assert_eq!(config.base().expire_batch(), 7, "the carry config must not detach a fresh base");
	}
}