reifydb-flow 0.9.1

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

use reifydb_codec::row::operator::state::{OperatorState, decode};
#[cfg(feature = "runtime")]
use reifydb_core::interface::catalog::flow::OperatorId;
use reifydb_core::{
	key::operator::state::{GroupId, GroupStateKey, KeyspaceId, OperatorStateKey},
	metrics::heap::HeapSize,
	state::timer::StateStore,
};
use reifydb_macro::operator_state;
use reifydb_value::{Result, value::datetime::DateTime};

#[cfg(feature = "runtime")]
use crate::transaction::{FlowTransaction, state::StateExtension};
use crate::{
	operator::state::seal::{coord::Coord, rule::SealedThrough},
	timer::Timer,
};

#[operator_state]
#[derive(Clone, Default)]
pub struct SealLedgerState {
	pub sealed_through: u64,
}

impl HeapSize for SealLedgerState {
	fn heap_size(&self) -> usize {
		0
	}
}

pub fn seal_ledger_key() -> GroupStateKey {
	OperatorStateKey::inner_encoded(GroupId::ROOT, KeyspaceId::SEAL_LEDGER, vec![])
}

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

impl FiredAt {
	pub fn of(timer: &Timer) -> Self {
		Self(timer.due)
	}

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

pub struct SealLedger;

impl SealLedger {
	pub fn read(store: &mut dyn StateStore) -> Result<Option<SealedThrough>> {
		Ok(Self::read_order(store)?.map(SealedThrough::from_order))
	}

	pub fn advance(store: &mut dyn StateStore, fired: FiredAt) -> Result<SealedThrough> {
		let fired_order = fired.at().to_order();
		let current = Self::read_order(store)?.unwrap_or(0);
		if fired_order <= current {
			return Ok(SealedThrough::from_order(current));
		}
		let state = SealLedgerState {
			sealed_through: fired_order,
		};
		store.state_set(&seal_ledger_key(), state.encode_state()?)?;
		Ok(SealedThrough::from_order(fired_order))
	}

	pub fn observe(store: &mut dyn StateStore, order: u64) -> Result<u64> {
		let current = Self::read_order(store)?.unwrap_or(0);
		if order <= current {
			return Ok(current);
		}
		let state = SealLedgerState {
			sealed_through: order,
		};
		store.state_set(&seal_ledger_key(), state.encode_state()?)?;
		Ok(order)
	}

	pub fn read_order(store: &mut dyn StateStore) -> Result<Option<u64>> {
		let Some(bytes) = store.state_get(&seal_ledger_key())? else {
			return Ok(None);
		};
		let state: SealLedgerState = decode(&bytes)?;
		Ok(Some(state.sealed_through))
	}
}

#[cfg(feature = "runtime")]
pub fn read_sealed_through<T: FlowTransaction>(txn: &mut T, operator: OperatorId) -> Result<Option<SealedThrough>> {
	let Some(row) = txn.state_get(operator, &seal_ledger_key())? else {
		return Ok(None);
	};
	let state: SealLedgerState = decode(&row)?;
	Ok(Some(SealedThrough::from_order(state.sealed_through)))
}

#[cfg(test)]
mod tests {
	use reifydb_codec::key::encoded::EncodedKey;
	use reifydb_core::state::timer::TimerKind;

	use super::*;
	use crate::operator::state::mock::MockStore;

	fn timer(millis: u64) -> Timer {
		Timer {
			due: DateTime::from_millis(millis),
			kind: TimerKind::Seal,
			key: EncodedKey::new(b"bucket".as_slice()),
		}
	}

	#[test]
	fn the_timer_seal_path_never_opens_to_arriving_data() {
		// FiredAt's only constructor takes a &Timer, so this path never seals on arriving data; observe may.
		let mut store = MockStore::default();

		let sealed = SealLedger::advance(&mut store, FiredAt::of(&timer(5_000))).unwrap();

		assert_eq!(sealed.at(), DateTime::from_millis(5_000));
	}

	#[test]
	fn two_timers_at_one_instant_seal_identically_whatever_key_they_carry() {
		// Seal entry points take a FiredAt, never a &Timer, so the timer's key cannot reach the
		// sealing decision. Both chaos and rolling arm seals with an empty key, so a sweep that
		// consulted timer.key would seal nothing for either.
		let keyed = Timer {
			due: DateTime::from_millis(5_000),
			kind: TimerKind::Seal,
			key: EncodedKey::new(b"some-window".as_slice()),
		};
		let keyless = Timer {
			due: DateTime::from_millis(5_000),
			kind: TimerKind::Seal,
			key: EncodedKey::new(Vec::new()),
		};

		assert_eq!(FiredAt::of(&keyed), FiredAt::of(&keyless));

		let mut keyed_store = MockStore::default();
		let mut keyless_store = MockStore::default();
		assert_eq!(
			SealLedger::advance(&mut keyed_store, FiredAt::of(&keyed)).unwrap(),
			SealLedger::advance(&mut keyless_store, FiredAt::of(&keyless)).unwrap()
		);
	}

	#[test]
	fn an_empty_ledger_reads_as_none_rather_than_the_epoch() {
		// none and "sealed through 1970" are different answers to reclaim: none means the operator has
		// no seal clamp, the epoch would mean everything is sealed and let reclaim erase live state.
		let mut store = MockStore::default();

		assert!(SealLedger::read(&mut store).unwrap().is_none());
	}

	#[test]
	fn the_ledger_only_moves_forward() {
		// Timers fire in (at, kind, key) order within a round, but a restart re-reads a cold wheel
		// and a late round can present an earlier instant. Rewinding the ledger would unclamp
		// reclaim and expose already-sealed windows to a second seal.
		let mut store = MockStore::default();

		SealLedger::advance(&mut store, FiredAt::of(&timer(9_000))).unwrap();
		let after_earlier = SealLedger::advance(&mut store, FiredAt::of(&timer(3_000))).unwrap();

		assert_eq!(after_earlier.at(), DateTime::from_millis(9_000));
		assert_eq!(SealLedger::read(&mut store).unwrap().unwrap().at(), DateTime::from_millis(9_000));
	}

	#[test]
	fn the_ledger_lives_in_the_root_group_and_carries_no_suffix() {
		// the key must be derivable from the operator alone, since reclaim reads it without knowing which group
		// wrote it
		let key = seal_ledger_key();
		let (group, keyspace, suffix) =
			OperatorStateKey::decode_inner(key.as_encoded().as_bytes()).expect("structured key");

		assert_eq!(group, GroupId::ROOT);
		assert_eq!(keyspace, KeyspaceId::SEAL_LEDGER);
		assert!(suffix.is_empty());
	}
}