reifydb_sub_server/

state.rs

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

use reifydb_auth::service::AuthService;
use reifydb_core::actors::server::ServerMessage;
use reifydb_engine::engine::StandardEngine;
use reifydb_runtime::{
	actor::{
		mailbox::ActorRef,
		system::{ActorHandle, ActorSpawner},
	},
	context::{clock::Clock, rng::Rng},
};
use reifydb_value::value::duration::Duration;
use tracing::instrument;

use crate::{actor::ServerActor, interceptor::RequestInterceptorChain};

#[derive(Debug, Clone)]
pub struct StateConfig {
	pub query_timeout: Duration,

	pub request_timeout: Duration,

	pub max_connections: usize,

	pub admin_enabled: bool,

	pub subscribe_max_hydration_rows: u64,

	pub subscribe_min_throttle: Duration,

	pub subscribe_min_linger: Duration,

	pub claim_wait_max: Duration,

	pub claim_lease_ttl: Duration,

	pub claim_max_parked: usize,
}

impl Default for StateConfig {
	fn default() -> Self {
		Self {
			query_timeout: Duration::from_seconds(30).unwrap(),
			request_timeout: Duration::from_seconds(60).unwrap(),
			max_connections: 10_000,
			admin_enabled: false,
			subscribe_max_hydration_rows: 10_000,
			subscribe_min_throttle: Duration::from_milliseconds(50).unwrap(),
			subscribe_min_linger: Duration::zero(),
			claim_wait_max: Duration::from_seconds(60).unwrap(),
			claim_lease_ttl: Duration::from_seconds(30).unwrap(),
			claim_max_parked: 10_000,
		}
	}
}

impl StateConfig {
	pub fn new() -> Self {
		Self::default()
	}

	pub fn query_timeout(mut self, timeout: Duration) -> Self {
		self.query_timeout = timeout;
		self
	}

	pub fn request_timeout(mut self, timeout: Duration) -> Self {
		self.request_timeout = timeout;
		self
	}

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

	pub fn admin_enabled(mut self, enabled: bool) -> Self {
		self.admin_enabled = enabled;
		self
	}

	pub fn subscribe_max_hydration_rows(mut self, n: u64) -> Self {
		self.subscribe_max_hydration_rows = n;
		self
	}

	pub fn subscribe_min_throttle(mut self, interval: Duration) -> Self {
		self.subscribe_min_throttle = interval;
		self
	}

	pub fn subscribe_min_linger(mut self, interval: Duration) -> Self {
		self.subscribe_min_linger = interval;
		self
	}

	pub fn claim_wait_max(mut self, budget: Duration) -> Self {
		self.claim_wait_max = budget;
		self
	}

	pub fn claim_lease_ttl(mut self, ttl: Duration) -> Self {
		self.claim_lease_ttl = ttl;
		self
	}

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

#[derive(Clone)]
pub struct AppState {
	spawner: ActorSpawner,
	engine: StandardEngine,
	auth_service: AuthService,
	config: StateConfig,
	request_interceptors: RequestInterceptorChain,
	clock: Clock,
	rng: Rng,
}

impl AppState {
	pub fn new(
		spawner: ActorSpawner,
		engine: StandardEngine,
		auth_service: AuthService,
		config: StateConfig,
		request_interceptors: RequestInterceptorChain,
		clock: Clock,
		rng: Rng,
	) -> Self {
		Self {
			spawner,
			engine,
			auth_service,
			config,
			request_interceptors,
			clock,
			rng,
		}
	}

	pub fn clone_with_config(&self, config: StateConfig) -> Self {
		Self {
			spawner: self.spawner.clone(),
			engine: self.engine.clone(),
			auth_service: self.auth_service.clone(),
			config,
			request_interceptors: self.request_interceptors.clone(),
			clock: self.clock.clone(),
			rng: self.rng.clone(),
		}
	}

	#[inline]
	pub fn spawner(&self) -> ActorSpawner {
		self.spawner.clone()
	}

	#[inline]
	pub fn engine(&self) -> &StandardEngine {
		&self.engine
	}

	#[inline]
	pub fn engine_clone(&self) -> StandardEngine {
		self.engine.clone()
	}

	#[inline]
	pub fn config(&self) -> &StateConfig {
		&self.config
	}

	#[inline]
	pub fn query_timeout(&self) -> Duration {
		self.config.query_timeout
	}

	#[inline]
	pub fn request_timeout(&self) -> Duration {
		self.config.request_timeout
	}

	#[inline]
	pub fn max_connections(&self) -> usize {
		self.config.max_connections
	}

	#[inline]
	pub fn admin_enabled(&self) -> bool {
		self.config.admin_enabled
	}

	#[inline]
	pub fn subscribe_max_hydration_rows(&self) -> u64 {
		self.config.subscribe_max_hydration_rows
	}

	#[inline]
	pub fn subscribe_min_throttle(&self) -> Duration {
		self.config.subscribe_min_throttle
	}

	#[inline]
	pub fn claim_wait_max(&self) -> Duration {
		self.config.claim_wait_max
	}

	#[inline]
	pub fn claim_lease_ttl(&self) -> Duration {
		self.config.claim_lease_ttl
	}

	#[inline]
	pub fn claim_max_parked(&self) -> usize {
		self.config.claim_max_parked
	}

	pub fn clamp_throttle(&self, requested: Option<Duration>) -> Duration {
		match requested {
			Some(d) if !d.is_zero() => d.max(self.config.subscribe_min_throttle),
			_ => Duration::zero(),
		}
	}

	#[inline]
	pub fn subscribe_min_linger(&self) -> Duration {
		self.config.subscribe_min_linger
	}

	pub fn clamp_linger(&self, requested: Option<Duration>) -> Duration {
		match requested {
			Some(d) if !d.is_zero() => d.max(self.config.subscribe_min_linger),
			_ => Duration::zero(),
		}
	}

	#[inline]
	pub fn request_interceptors(&self) -> &RequestInterceptorChain {
		&self.request_interceptors
	}

	#[inline]
	pub fn clock(&self) -> &Clock {
		&self.clock
	}

	#[inline]
	pub fn rng(&self) -> &Rng {
		&self.rng
	}

	#[inline]
	pub fn auth_service(&self) -> &AuthService {
		&self.auth_service
	}

	#[instrument(name = "actor::spawn_server", level = "info", skip_all)]
	pub fn spawn_server_actor(&self) -> (ActorRef<ServerMessage>, ActorHandle<ServerMessage>) {
		let actor = ServerActor::new(self.engine.clone(), self.auth_service.clone(), self.clock.clone());
		let handle = self.spawner.spawn_ephemeral("server-req", actor);
		let actor_ref = handle.actor_ref().clone();
		(actor_ref, handle)
	}
}

#[cfg(test)]
pub mod tests {
	use super::*;

	#[test]
	fn test_query_defaults() {
		let config = StateConfig::default();
		assert_eq!(config.query_timeout, Duration::from_seconds(30).unwrap());
		assert_eq!(config.request_timeout, Duration::from_seconds(60).unwrap());
		assert_eq!(config.max_connections, 10_000);
		assert_eq!(config.subscribe_max_hydration_rows, 10_000);
		assert_eq!(config.subscribe_min_throttle, Duration::from_milliseconds(50).unwrap());
		assert_eq!(config.subscribe_min_linger, Duration::zero());
	}

	#[test]
	fn test_query_config_builder() {
		let config = StateConfig::new()
			.query_timeout(Duration::from_seconds(60).unwrap())
			.request_timeout(Duration::from_seconds(120).unwrap())
			.max_connections(5_000)
			.subscribe_max_hydration_rows(50_000)
			.subscribe_min_throttle(Duration::from_milliseconds(200).unwrap())
			.subscribe_min_linger(Duration::from_milliseconds(20).unwrap());

		assert_eq!(config.query_timeout, Duration::from_seconds(60).unwrap());
		assert_eq!(config.request_timeout, Duration::from_seconds(120).unwrap());
		assert_eq!(config.max_connections, 5_000);
		assert_eq!(config.subscribe_max_hydration_rows, 50_000);
		assert_eq!(config.subscribe_min_throttle, Duration::from_milliseconds(200).unwrap());
		assert_eq!(config.subscribe_min_linger, Duration::from_milliseconds(20).unwrap());
	}
}