reifydb-runtime 0.10.1

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

#![allow(clippy::disallowed_types)]

mod pool;

use std::{
	any::Any,
	error, fmt,
	fmt::{Debug, Formatter},
	mem,
	sync::{Arc, OnceLock, Weak},
	time,
	time::Duration,
};

use crossbeam_channel::{Receiver, RecvTimeoutError as CcRecvTimeoutError};

use crate::{
	actor::{
		context::CancellationToken, system::host::pool::PoolActorHandle, timers::scheduler::SchedulerHandle,
		traits::Actor,
	},
	context::clock::Clock,
	pool::{
		PoolConfig, Pools,
		actor_pool::{EPHEMERAL_BATCH_SIZE, Schedule},
	},
	sync::mutex::Mutex,
};

static TESTING_ROOT: OnceLock<ActorSystem> = OnceLock::new();

struct ActorSystemInner {
	cancel: CancellationToken,
	scheduler: SchedulerHandle,
	clock: Clock,
	pools: Pools,
	wakers: Mutex<Vec<Arc<dyn Fn() + Send + Sync>>>,
	keepalive: Mutex<Vec<Box<dyn Any + Send + Sync>>>,
	done_rxs: Mutex<Vec<Receiver<Result<(), String>>>>,
	children: Mutex<Vec<ActorSystem>>,
	parent: Weak<ActorSystemInner>,
}

#[derive(Clone)]
pub struct ActorSystem {
	inner: Arc<ActorSystemInner>,
}

impl ActorSystem {
	pub fn new(pools: Pools, clock: Clock) -> Self {
		let scheduler = SchedulerHandle::new();

		Self {
			inner: Arc::new(ActorSystemInner {
				cancel: CancellationToken::new(),
				scheduler,
				clock,
				pools,
				wakers: Mutex::new(Vec::new()),
				keepalive: Mutex::new(Vec::new()),
				done_rxs: Mutex::new(Vec::new()),
				children: Mutex::new(Vec::new()),
				parent: Weak::new(),
			}),
		}
	}

	pub fn testing(clock: Clock) -> Self {
		TESTING_ROOT.get_or_init(|| Self::new(Pools::new(PoolConfig::default()), Clock::Real)).scope_with(clock)
	}

	pub fn scope(&self) -> Self {
		self.scope_with(self.inner.clock.clone())
	}

	fn scope_with(&self, clock: Clock) -> Self {
		let cancel = self.inner.cancel.child_token();
		let child = Self {
			inner: Arc::new(ActorSystemInner {
				scheduler: self.inner.scheduler.shared(cancel.clone()),
				cancel,
				clock,
				pools: self.inner.pools.clone(),
				wakers: Mutex::new(Vec::new()),
				keepalive: Mutex::new(Vec::new()),
				done_rxs: Mutex::new(Vec::new()),
				children: Mutex::new(Vec::new()),
				parent: Arc::downgrade(&self.inner),
			}),
		};
		self.inner.children.lock().push(child.clone());
		child
	}

	pub fn pools(&self) -> Pools {
		self.inner.pools.clone()
	}

	pub fn spawner(&self) -> ActorSpawner {
		ActorSpawner {
			inner: Arc::downgrade(&self.inner),
			clock: self.inner.clock.clone(),
		}
	}

	pub fn cancellation_token(&self) -> CancellationToken {
		self.inner.cancel.clone()
	}

	pub fn is_cancelled(&self) -> bool {
		self.inner.cancel.is_cancelled()
	}

	pub fn shutdown(&self) {
		self.inner.cancel.cancel();

		let children = mem::take(&mut *self.inner.children.lock());
		for child in &children {
			child.shutdown();
		}
		drop(children);

		let wakers = mem::take(&mut *self.inner.wakers.lock());
		for waker in &wakers {
			waker();
		}
		drop(wakers);

		self.inner.scheduler.shutdown();

		self.inner.keepalive.lock().clear();

		if let Some(parent) = self.inner.parent.upgrade() {
			parent.children.lock().retain(|child| !Arc::ptr_eq(&child.inner, &self.inner));
		}
	}

	pub(crate) fn register_waker(&self, f: Arc<dyn Fn() + Send + Sync>) {
		self.inner.wakers.lock().push(f);
	}

	pub(crate) fn register_keepalive(&self, cell: Box<dyn Any + Send + Sync>) {
		self.inner.keepalive.lock().push(cell);
	}

	pub(crate) fn register_done_rx(&self, rx: Receiver<Result<(), String>>) {
		self.inner.done_rxs.lock().push(rx);
	}

	pub fn join(&self) -> Result<(), JoinError> {
		self.join_timeout(Duration::from_secs(5))
	}

	#[allow(clippy::disallowed_methods)]
	pub fn join_timeout(&self, timeout: Duration) -> Result<(), JoinError> {
		let deadline = time::Instant::now() + timeout;
		let rxs: Vec<_> = mem::take(&mut *self.inner.done_rxs.lock());
		for rx in rxs {
			let remaining = deadline.saturating_duration_since(time::Instant::now());
			match rx.recv_timeout(remaining) {
				Ok(Ok(())) => {}
				Ok(Err(name)) => {
					return Err(JoinError::new(format!("actor {name} panicked")));
				}
				Err(CcRecvTimeoutError::Disconnected) => {}
				Err(CcRecvTimeoutError::Timeout) => {
					return Err(JoinError::new("timed out waiting for actors to stop"));
				}
			}
		}
		Ok(())
	}

	pub fn scheduler(&self) -> &SchedulerHandle {
		&self.inner.scheduler
	}

	pub fn clock(&self) -> &Clock {
		&self.inner.clock
	}

	pub fn spawn_coordination<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		let group = self.inner.pools.actor_pool().coordination();
		pool::spawn_on_schedule(self, name, actor, Schedule::Pinned(group.assign()), group.batch_size())
	}

	pub fn spawn_flow<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		let group = self.inner.pools.actor_pool().flow();
		pool::spawn_on_schedule(self, name, actor, Schedule::Pinned(group.assign()), group.batch_size())
	}

	pub fn spawn_maintenance<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		let group = self.inner.pools.actor_pool().maintenance();
		pool::spawn_on_schedule(self, name, actor, Schedule::Pinned(group.assign()), group.batch_size())
	}

	pub fn spawn_ephemeral<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		pool::spawn_on_schedule(self, name, actor, self.inner.pools.task_injector(), EPHEMERAL_BATCH_SIZE)
	}
}

impl Debug for ActorSystem {
	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
		f.debug_struct("ActorSystem").field("cancelled", &self.is_cancelled()).finish_non_exhaustive()
	}
}

#[derive(Clone)]
pub struct ActorSpawner {
	inner: Weak<ActorSystemInner>,
	clock: Clock,
}

impl ActorSpawner {
	fn system(&self) -> ActorSystem {
		ActorSystem {
			inner: self.inner.upgrade().expect("runtime already shut down: cannot spawn actor"),
		}
	}

	pub fn clock(&self) -> &Clock {
		&self.clock
	}

	pub fn pools(&self) -> Pools {
		self.system().pools()
	}

	pub fn is_alive(&self) -> bool {
		self.inner.strong_count() > 0
	}

	pub fn cancellation_token(&self) -> Option<CancellationToken> {
		self.inner.upgrade().map(|inner| inner.cancel.clone())
	}

	pub fn scope(&self) -> ActorSpawner {
		self.system().scope().spawner()
	}

	pub fn shutdown(&self) {
		if let Some(inner) = self.inner.upgrade() {
			ActorSystem {
				inner,
			}
			.shutdown();
		}
	}

	pub fn spawn_coordination<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		self.system().spawn_coordination(name, actor)
	}

	pub fn spawn_flow<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		self.system().spawn_flow(name, actor)
	}

	pub fn spawn_maintenance<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		self.system().spawn_maintenance(name, actor)
	}

	pub fn spawn_ephemeral<A: Actor>(&self, name: &str, actor: A) -> ActorHandle<A::Message>
	where
		A::State: Send,
	{
		self.system().spawn_ephemeral(name, actor)
	}
}

impl Debug for ActorSpawner {
	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
		f.debug_struct("ActorSpawner").field("alive", &self.is_alive()).finish_non_exhaustive()
	}
}

pub type ActorHandle<M> = PoolActorHandle<M>;

#[derive(Debug)]
pub struct JoinError {
	message: String,
}

impl JoinError {
	pub fn new(message: impl Into<String>) -> Self {
		Self {
			message: message.into(),
		}
	}
}

impl fmt::Display for JoinError {
	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
		write!(f, "actor join failed: {}", self.message)
	}
}

impl error::Error for JoinError {}

#[cfg(test)]
mod tests {
	use std::sync;

	use super::*;
	use crate::{
		actor::{context::Context, mailbox::ActorRef, system::ActorConfig, traits::Directive},
		pool::{PoolConfig, Pools},
	};

	fn test_system() -> ActorSystem {
		let pools = Pools::new(PoolConfig::default());
		ActorSystem::new(pools, Clock::Real)
	}

	struct CounterActor;

	#[derive(Debug)]
	enum CounterMessage {
		Inc,
		Get(sync::mpsc::Sender<i64>),
		Stop,
	}

	impl Actor for CounterActor {
		type State = i64;
		type Message = CounterMessage;

		fn init(&self, _ctx: &Context<Self::Message>) -> Self::State {
			0
		}

		fn handle(
			&self,
			state: &mut Self::State,
			msg: Self::Message,
			_ctx: &Context<Self::Message>,
		) -> Directive {
			match msg {
				CounterMessage::Inc => *state += 1,
				CounterMessage::Get(tx) => {
					let _ = tx.send(*state);
				}
				CounterMessage::Stop => return Directive::Stop,
			}
			Directive::Continue
		}
	}

	#[test]
	fn test_spawn_and_send() {
		let system = test_system();
		let handle = system.spawn_coordination("counter", CounterActor);

		let actor_ref = handle.actor_ref().clone();
		actor_ref.send(CounterMessage::Inc).unwrap();
		actor_ref.send(CounterMessage::Inc).unwrap();
		actor_ref.send(CounterMessage::Inc).unwrap();

		let (tx, rx) = sync::mpsc::channel();
		actor_ref.send(CounterMessage::Get(tx)).unwrap();

		let value = rx.recv().unwrap();
		assert_eq!(value, 3);

		actor_ref.send(CounterMessage::Stop).unwrap();
		handle.join().unwrap();
	}

	#[test]
	fn test_shutdown_join() {
		let system = test_system();

		for i in 0..5 {
			system.spawn_coordination(&format!("counter-{i}"), CounterActor);
		}

		// join() must not return before every actor has finished, or shutdown races teardown.
		system.shutdown();
		system.join().unwrap();
	}

	#[test]
	fn test_shutdown_stops_the_timer_scheduler() {
		// Timers must be dead before the cells are released, otherwise a callback owns the last system ref.
		let system = test_system();
		system.shutdown();

		let (tx, rx) = sync::mpsc::channel();
		system.scheduler().schedule_once(Duration::from_millis(5), move || {
			let _ = tx.send(());
		});

		assert!(rx.recv_timeout(Duration::from_millis(200)).is_err());
	}

	#[test]
	fn a_scope_that_shuts_down_leaves_its_parent_but_a_live_scope_still_stops_with_it() {
		// A parent that keeps every stopped scope grows by one system per open and close, forever.
		let root = test_system();
		let stopped = root.scope();
		let live = root.scope();
		assert_eq!(root.inner.children.lock().len(), 2, "precondition: both scopes are registered");

		stopped.shutdown();
		let children = root.inner.children.lock().clone();
		assert_eq!(children.len(), 1, "a stopped scope must leave its parent's list");
		assert!(Arc::ptr_eq(&children[0].inner, &live.inner), "only the stopped scope may leave");
		drop(children);

		root.shutdown();
		assert!(live.is_cancelled(), "a parent shutdown must still stop every live scope");
		assert!(root.inner.children.lock().is_empty(), "the parent must hold no scope after its shutdown");
	}
	struct TurnActor {
		log: Arc<sync::Mutex<Vec<&'static str>>>,
		name: &'static str,
		batch: Option<usize>,
	}

	#[derive(Debug)]
	enum TurnMessage {
		Hold(sync::mpsc::Sender<()>, sync::mpsc::Receiver<()>),
		Kick(ActorRef<TurnMessage>),
		Log,
		Done(sync::mpsc::Sender<()>),
	}

	impl Actor for TurnActor {
		type State = ();
		type Message = TurnMessage;

		fn init(&self, _ctx: &Context<Self::Message>) -> Self::State {}

		fn handle(
			&self,
			_state: &mut Self::State,
			msg: Self::Message,
			_ctx: &Context<Self::Message>,
		) -> Directive {
			match msg {
				TurnMessage::Hold(started, release) => {
					let _ = started.send(());
					let _ = release.recv();
				}
				TurnMessage::Kick(other) => {
					let _ = other.send(TurnMessage::Log);
				}
				TurnMessage::Log => self.log.lock().unwrap().push(self.name),
				TurnMessage::Done(tx) => {
					let _ = tx.send(());
				}
			}
			Directive::Continue
		}

		fn config(&self) -> ActorConfig {
			match self.batch {
				Some(batch) => ActorConfig::new().batch_size(batch),
				None => ActorConfig::new(),
			}
		}
	}

	fn turn_order(batch: Option<usize>) -> Vec<&'static str> {
		let system = ActorSystem::new(
			Pools::new(PoolConfig {
				flow_threads: 1,
				..PoolConfig::default()
			}),
			Clock::Real,
		);
		let log = Arc::new(sync::Mutex::new(Vec::new()));
		let actor = |name, batch| TurnActor {
			log: Arc::clone(&log),
			name,
			batch,
		};
		let holder = system.spawn_flow("holder", actor("holder", None));
		let busy = system.spawn_flow("busy", actor("busy", batch));
		let other = system.spawn_flow("other", actor("other", None));

		for handle in [&busy, &other] {
			let (tx, rx) = sync::mpsc::channel();
			handle.actor_ref().send(TurnMessage::Done(tx)).unwrap();
			rx.recv().unwrap();
		}

		let (started_tx, started_rx) = sync::mpsc::channel();
		let (release_tx, release_rx) = sync::mpsc::channel();
		holder.actor_ref().send(TurnMessage::Hold(started_tx, release_rx)).unwrap();
		started_rx.recv().unwrap();

		busy.actor_ref().send(TurnMessage::Kick(other.actor_ref().clone())).unwrap();
		for _ in 0..20 {
			busy.actor_ref().send(TurnMessage::Log).unwrap();
		}
		let (busy_done_tx, busy_done_rx) = sync::mpsc::channel();
		busy.actor_ref().send(TurnMessage::Done(busy_done_tx)).unwrap();
		release_tx.send(()).unwrap();
		busy_done_rx.recv().unwrap();

		let (other_done_tx, other_done_rx) = sync::mpsc::channel();
		other.actor_ref().send(TurnMessage::Done(other_done_tx)).unwrap();
		other_done_rx.recv().unwrap();

		system.shutdown();
		system.join().unwrap();
		log.lock().unwrap().clone()
	}

	#[test]
	fn an_actor_batch_size_holds_its_turn_for_that_many_messages() {
		// A shared queue that yields after the pool default makes every sender wait behind the whole pool.
		let own = turn_order(Some(64));
		assert_eq!(
			own.iter().position(|name| *name == "other"),
			Some(20),
			"a batch of 64 must finish all 20 queued messages before a newly woken actor runs: {own:?}"
		);

		let pooled = turn_order(None);
		assert_eq!(
			pooled.iter().position(|name| *name == "other"),
			Some(7),
			"without its own batch size the actor must yield after the flow pool default of 8: {pooled:?}"
		);
	}

	struct PanicActor;

	impl Actor for PanicActor {
		type State = ();
		type Message = sync::mpsc::Sender<()>;

		fn init(&self, _ctx: &Context<Self::Message>) -> Self::State {}

		fn handle(
			&self,
			_state: &mut Self::State,
			reached: Self::Message,
			_ctx: &Context<Self::Message>,
		) -> Directive {
			let _ = reached.send(());
			panic!("a pool actor crashed while handling a message");
		}
	}

	#[test]
	fn a_pool_actor_that_panics_fails_the_join_instead_of_passing_shutdown_as_clean() {
		// A join that reads a dropped done signal as a clean stop hides a crashed actor behind a clean
		// shutdown.
		let system = test_system();
		let crashed = system.spawn_flow("crashed", PanicActor);
		let (reached_tx, reached_rx) = sync::mpsc::channel();
		crashed.actor_ref().send(reached_tx).unwrap();
		reached_rx.recv().unwrap();

		system.shutdown();

		assert!(system.join().is_err(), "a crashed pool actor must fail the join");
	}

	#[test]
	fn a_pool_actor_that_panics_fails_its_handle_join_after_shutdown() {
		// Subsystems join each actor handle on shutdown; a crashed actor read as joined would report a clean
		// stop.
		let system = test_system();
		let crashed = system.spawn_flow("crashed", PanicActor);
		let (reached_tx, reached_rx) = sync::mpsc::channel();
		crashed.actor_ref().send(reached_tx).unwrap();
		reached_rx.recv().unwrap();

		system.shutdown();

		assert!(crashed.join().is_err(), "a crashed pool actor must fail its handle join");
	}
}