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());
}
}