reifydb_sub_server/
state.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2025 ReifyDB
3
4use std::time::Duration;
5
6use reifydb_auth::service::AuthService;
7use reifydb_core::actors::server::ServerMessage;
8use reifydb_engine::engine::StandardEngine;
9use reifydb_runtime::{
10	actor::{
11		mailbox::ActorRef,
12		system::{ActorHandle, ActorSystem},
13	},
14	context::{clock::Clock, rng::Rng},
15};
16use tracing::instrument;
17
18use crate::{actor::ServerActor, interceptor::RequestInterceptorChain};
19
20#[derive(Debug, Clone)]
21pub struct StateConfig {
22	pub query_timeout: Duration,
23
24	pub request_timeout: Duration,
25
26	pub max_connections: usize,
27
28	pub admin_enabled: bool,
29
30	pub subscribe_max_hydration_rows: u64,
31}
32
33impl Default for StateConfig {
34	fn default() -> Self {
35		Self {
36			query_timeout: Duration::from_secs(30),
37			request_timeout: Duration::from_secs(60),
38			max_connections: 10_000,
39			admin_enabled: false,
40			subscribe_max_hydration_rows: 10_000,
41		}
42	}
43}
44
45impl StateConfig {
46	pub fn new() -> Self {
47		Self::default()
48	}
49
50	pub fn query_timeout(mut self, timeout: Duration) -> Self {
51		self.query_timeout = timeout;
52		self
53	}
54
55	pub fn request_timeout(mut self, timeout: Duration) -> Self {
56		self.request_timeout = timeout;
57		self
58	}
59
60	pub fn max_connections(mut self, max: usize) -> Self {
61		self.max_connections = max;
62		self
63	}
64
65	pub fn admin_enabled(mut self, enabled: bool) -> Self {
66		self.admin_enabled = enabled;
67		self
68	}
69
70	pub fn subscribe_max_hydration_rows(mut self, n: u64) -> Self {
71		self.subscribe_max_hydration_rows = n;
72		self
73	}
74}
75
76#[derive(Clone)]
77pub struct AppState {
78	actor_system: ActorSystem,
79	engine: StandardEngine,
80	auth_service: AuthService,
81	config: StateConfig,
82	request_interceptors: RequestInterceptorChain,
83	clock: Clock,
84	rng: Rng,
85}
86
87impl AppState {
88	pub fn new(
89		actor_system: ActorSystem,
90		engine: StandardEngine,
91		auth_service: AuthService,
92		config: StateConfig,
93		request_interceptors: RequestInterceptorChain,
94		clock: Clock,
95		rng: Rng,
96	) -> Self {
97		Self {
98			actor_system,
99			engine,
100			auth_service,
101			config,
102			request_interceptors,
103			clock,
104			rng,
105		}
106	}
107
108	pub fn clone_with_config(&self, config: StateConfig) -> Self {
109		Self {
110			actor_system: self.actor_system.clone(),
111			engine: self.engine.clone(),
112			auth_service: self.auth_service.clone(),
113			config,
114			request_interceptors: self.request_interceptors.clone(),
115			clock: self.clock.clone(),
116			rng: self.rng.clone(),
117		}
118	}
119
120	#[inline]
121	pub fn actor_system(&self) -> ActorSystem {
122		self.actor_system.clone()
123	}
124
125	#[inline]
126	pub fn engine(&self) -> &StandardEngine {
127		&self.engine
128	}
129
130	#[inline]
131	pub fn engine_clone(&self) -> StandardEngine {
132		self.engine.clone()
133	}
134
135	#[inline]
136	pub fn config(&self) -> &StateConfig {
137		&self.config
138	}
139
140	#[inline]
141	pub fn query_timeout(&self) -> Duration {
142		self.config.query_timeout
143	}
144
145	#[inline]
146	pub fn request_timeout(&self) -> Duration {
147		self.config.request_timeout
148	}
149
150	#[inline]
151	pub fn max_connections(&self) -> usize {
152		self.config.max_connections
153	}
154
155	#[inline]
156	pub fn admin_enabled(&self) -> bool {
157		self.config.admin_enabled
158	}
159
160	#[inline]
161	pub fn subscribe_max_hydration_rows(&self) -> u64 {
162		self.config.subscribe_max_hydration_rows
163	}
164
165	#[inline]
166	pub fn request_interceptors(&self) -> &RequestInterceptorChain {
167		&self.request_interceptors
168	}
169
170	#[inline]
171	pub fn clock(&self) -> &Clock {
172		&self.clock
173	}
174
175	#[inline]
176	pub fn rng(&self) -> &Rng {
177		&self.rng
178	}
179
180	#[inline]
181	pub fn auth_service(&self) -> &AuthService {
182		&self.auth_service
183	}
184
185	#[instrument(name = "actor::spawn_server", level = "debug", skip_all)]
186	pub fn spawn_server_actor(&self) -> (ActorRef<ServerMessage>, ActorHandle<ServerMessage>) {
187		let actor = ServerActor::new(self.engine.clone(), self.auth_service.clone(), self.clock.clone());
188		let handle = self.actor_system.spawn_query("server-req", actor);
189		let actor_ref = handle.actor_ref().clone();
190		(actor_ref, handle)
191	}
192}
193
194#[cfg(test)]
195pub mod tests {
196	use super::*;
197
198	#[test]
199	fn test_query_defaults() {
200		let config = StateConfig::default();
201		assert_eq!(config.query_timeout, Duration::from_secs(30));
202		assert_eq!(config.request_timeout, Duration::from_secs(60));
203		assert_eq!(config.max_connections, 10_000);
204		assert_eq!(config.subscribe_max_hydration_rows, 10_000);
205	}
206
207	#[test]
208	fn test_query_config_builder() {
209		let config = StateConfig::new()
210			.query_timeout(Duration::from_secs(60))
211			.request_timeout(Duration::from_secs(120))
212			.max_connections(5_000)
213			.subscribe_max_hydration_rows(50_000);
214
215		assert_eq!(config.query_timeout, Duration::from_secs(60));
216		assert_eq!(config.request_timeout, Duration::from_secs(120));
217		assert_eq!(config.max_connections, 5_000);
218		assert_eq!(config.subscribe_max_hydration_rows, 50_000);
219	}
220}
reifydb_sub_server/state.rs

reifydb_sub_server/
state.rs
