1use reifydb_auth::service::AuthService;
5use reifydb_core::util::ioc::IocContainer;
6use reifydb_engine::engine::StandardEngine;
7use reifydb_runtime::{
8 actor::system::ActorSpawner,
9 context::{clock::Clock, rng::Rng},
10};
11use reifydb_sub_api::subsystem::{Subsystem, SubsystemFactory};
12use reifydb_sub_server::{
13 interceptor::RequestInterceptorChain,
14 state::{AppState, StateConfig},
15};
16use reifydb_value::{Result, value::duration::Duration};
17use tokio::runtime::Handle;
18
19use crate::subsystem::HttpSubsystem;
20
21pub struct HttpConfigurator {
22 bind_addr: Option<String>,
23 admin_bind_addr: Option<String>,
24 max_connections: usize,
25 query_timeout: Duration,
26 request_timeout: Duration,
27 claim_wait_max: Duration,
28 claim_max_parked: usize,
29 spawner: Option<ActorSpawner>,
30 clock: Option<Clock>,
31 rng: Option<Rng>,
32 handle: Option<Handle>,
33}
34
35impl Default for HttpConfigurator {
36 fn default() -> Self {
37 Self::new()
38 }
39}
40
41impl HttpConfigurator {
42 pub fn new() -> Self {
43 Self {
44 bind_addr: None,
45 admin_bind_addr: None,
46 max_connections: 10_000,
47 query_timeout: Duration::from_seconds(30).unwrap(),
48 request_timeout: Duration::from_seconds(60).unwrap(),
49 claim_wait_max: Duration::from_seconds(60).unwrap(),
50 claim_max_parked: 10_000,
51 spawner: None,
52 clock: None,
53 rng: None,
54 handle: None,
55 }
56 }
57
58 pub fn bind_addr(mut self, addr: impl Into<String>) -> Self {
59 self.bind_addr = Some(addr.into());
60 self
61 }
62
63 pub fn admin_bind_addr(mut self, addr: impl Into<String>) -> Self {
64 self.admin_bind_addr = Some(addr.into());
65 self
66 }
67
68 pub fn max_connections(mut self, max: usize) -> Self {
69 self.max_connections = max;
70 self
71 }
72
73 pub fn claim_wait_max(mut self, budget: Duration) -> Self {
74 self.claim_wait_max = budget;
75 self
76 }
77
78 pub fn claim_max_parked(mut self, max: usize) -> Self {
79 self.claim_max_parked = max;
80 self
81 }
82
83 pub fn query_timeout(mut self, timeout: Duration) -> Self {
84 self.query_timeout = timeout;
85 self
86 }
87
88 pub fn request_timeout(mut self, timeout: Duration) -> Self {
89 self.request_timeout = timeout;
90 self
91 }
92
93 pub fn spawner(mut self, spawner: ActorSpawner) -> Self {
94 self.spawner = Some(spawner);
95 self
96 }
97
98 pub fn clock(mut self, clock: Clock) -> Self {
99 self.clock = Some(clock);
100 self
101 }
102
103 pub fn rng(mut self, rng: Rng) -> Self {
104 self.rng = Some(rng);
105 self
106 }
107
108 pub fn handle(mut self, handle: Handle) -> Self {
109 self.handle = Some(handle);
110 self
111 }
112
113 pub(crate) fn configure(self) -> HttpConfig {
114 HttpConfig {
115 bind_addr: self.bind_addr,
116 admin_bind_addr: self.admin_bind_addr,
117 max_connections: self.max_connections,
118 query_timeout: self.query_timeout,
119 request_timeout: self.request_timeout,
120 claim_wait_max: self.claim_wait_max,
121 claim_max_parked: self.claim_max_parked,
122 spawner: self.spawner,
123 clock: self.clock,
124 rng: self.rng,
125 handle: self.handle,
126 }
127 }
128}
129
130#[derive(Clone)]
131pub struct HttpConfig {
132 pub bind_addr: Option<String>,
133
134 pub admin_bind_addr: Option<String>,
135
136 pub max_connections: usize,
137
138 pub query_timeout: Duration,
139
140 pub request_timeout: Duration,
141
142 pub claim_wait_max: Duration,
143
144 pub claim_max_parked: usize,
145
146 pub spawner: Option<ActorSpawner>,
147
148 pub clock: Option<Clock>,
149
150 pub rng: Option<Rng>,
151
152 pub handle: Option<Handle>,
153}
154
155impl Default for HttpConfig {
156 fn default() -> Self {
157 HttpConfigurator::new().configure()
158 }
159}
160
161pub struct HttpSubsystemFactory {
162 config_fn: Box<dyn FnOnce() -> HttpConfig + Send>,
163}
164
165impl HttpSubsystemFactory {
166 pub fn new<F>(configurator: F) -> Self
167 where
168 F: FnOnce(HttpConfigurator) -> HttpConfigurator + Send + 'static,
169 {
170 Self {
171 config_fn: Box::new(move || configurator(HttpConfigurator::new()).configure()),
172 }
173 }
174}
175
176type ResolvedDeps = (StandardEngine, RequestInterceptorChain, ActorSpawner, Clock, Rng, Handle, AuthService);
177
178impl SubsystemFactory for HttpSubsystemFactory {
179 fn create(self: Box<Self>, ioc: &IocContainer) -> Result<Box<dyn Subsystem>> {
180 let config = (self.config_fn)();
181
182 let (engine, interceptors, spawner, clock, rng, handle, auth_service) =
183 Self::resolve_deps(ioc, config.spawner, config.clock, config.rng, config.handle)?;
184
185 let state = Self::build_app_state(
186 config.query_timeout,
187 config.request_timeout,
188 config.max_connections,
189 config.claim_wait_max,
190 config.claim_max_parked,
191 spawner,
192 engine,
193 clock,
194 rng,
195 interceptors,
196 auth_service,
197 );
198
199 let subsystem =
200 Self::build_subsystem(config.bind_addr.clone(), config.admin_bind_addr.clone(), state, handle)?;
201
202 Ok(Box::new(subsystem))
203 }
204}
205
206impl HttpSubsystemFactory {
207 #[inline]
208 #[allow(clippy::too_many_arguments)]
209 fn resolve_deps(
210 ioc: &IocContainer,
211 spawner: Option<ActorSpawner>,
212 clock: Option<Clock>,
213 rng: Option<Rng>,
214 handle: Option<Handle>,
215 ) -> Result<ResolvedDeps> {
216 let engine = ioc.resolve::<StandardEngine>()?;
217 let interceptors = ioc.resolve::<RequestInterceptorChain>().unwrap_or_default();
218
219 let spawner = match spawner {
220 Some(spawner) => spawner,
221 None => ioc.resolve::<ActorSpawner>()?,
222 };
223 let clock = match clock {
224 Some(clock) => clock,
225 None => ioc.resolve::<Clock>()?,
226 };
227 let rng = match rng {
228 Some(rng) => rng,
229 None => ioc.resolve::<Rng>()?,
230 };
231 let handle = match handle {
232 Some(handle) => handle,
233 None => ioc.resolve::<Handle>()?,
234 };
235
236 let auth_service = ioc.resolve::<AuthService>()?;
237
238 Ok((engine, interceptors, spawner, clock, rng, handle, auth_service))
239 }
240
241 #[inline]
242 #[allow(clippy::too_many_arguments)]
243 fn build_app_state(
244 query_timeout: Duration,
245 request_timeout: Duration,
246 max_connections: usize,
247 claim_wait_max: Duration,
248 claim_max_parked: usize,
249 spawner: ActorSpawner,
250 engine: StandardEngine,
251 clock: Clock,
252 rng: Rng,
253 interceptors: RequestInterceptorChain,
254 auth_service: AuthService,
255 ) -> AppState {
256 let query_config = StateConfig::new()
257 .query_timeout(query_timeout)
258 .request_timeout(request_timeout)
259 .max_connections(max_connections)
260 .claim_wait_max(claim_wait_max)
261 .claim_max_parked(claim_max_parked);
262
263 AppState::new(spawner, engine, auth_service, query_config, interceptors, clock, rng)
264 }
265
266 #[inline]
267 fn build_subsystem(
268 bind_addr: Option<String>,
269 admin_bind_addr: Option<String>,
270 state: AppState,
271 handle: Handle,
272 ) -> Result<HttpSubsystem> {
273 HttpSubsystem::new(bind_addr, admin_bind_addr, state, handle)
274 }
275}