1use std::ffi::{OsStr, OsString};
2use std::net::SocketAddr;
3use std::path::PathBuf;
4
5use crate::ServerError;
6
7use super::types::ServerConfig;
8
9const ENV_PREFIX: &str = "LIMINAL_";
10const LISTEN_ADDRESS: &str = "LIMINAL_LISTEN_ADDRESS";
11const HEALTH_LISTEN_ADDRESS: &str = "LIMINAL_HEALTH_LISTEN_ADDRESS";
12const DRAIN_TIMEOUT_MS: &str = "LIMINAL_DRAIN_TIMEOUT_MS";
13const PERSISTENCE_PATH: &str = "LIMINAL_PERSISTENCE_PATH";
14const CLUSTER_NODE_NAME: &str = "LIMINAL_CLUSTER_NODE_NAME";
15const CLUSTER_SEED_NODES: &str = "LIMINAL_CLUSTER_SEED_NODES";
16const CLUSTER_LISTEN_ADDRESS: &str = "LIMINAL_CLUSTER_LISTEN_ADDRESS";
17const CLUSTER_COOKIE: &str = "LIMINAL_CLUSTER_COOKIE";
18const AUTH_TOKEN: &str = "LIMINAL_AUTH_TOKEN";
19
20pub fn apply_env_overrides(config: ServerConfig) -> Result<ServerConfig, ServerError> {
43 apply_env_overrides_from(config, std::env::vars_os())
44}
45
46pub(crate) fn apply_env_overrides_from<I>(
47 mut config: ServerConfig,
48 variables: I,
49) -> Result<ServerConfig, ServerError>
50where
51 I: IntoIterator<Item = (OsString, OsString)>,
52{
53 for (key, value) in variables {
54 let Some(key) = key.to_str() else {
55 continue;
56 };
57
58 if !key.starts_with(ENV_PREFIX) {
59 continue;
60 }
61
62 match key {
63 LISTEN_ADDRESS => {
64 config.listen_address = parse_socket_addr(LISTEN_ADDRESS, &value)?;
65 }
66 HEALTH_LISTEN_ADDRESS => {
67 config.health_listen_address = parse_socket_addr(HEALTH_LISTEN_ADDRESS, &value)?;
68 }
69 DRAIN_TIMEOUT_MS => {
70 config.drain_timeout_ms = parse_u64(DRAIN_TIMEOUT_MS, &value)?;
71 }
72 PERSISTENCE_PATH => {
73 config.persistence_path = Some(PathBuf::from(value));
74 }
75 CLUSTER_NODE_NAME => {
76 let node_name = env_string(CLUSTER_NODE_NAME, &value)?;
77 cluster_required(&mut config, CLUSTER_NODE_NAME)?.node_name = node_name;
78 }
79 CLUSTER_SEED_NODES => {
80 let seed_nodes = parse_seed_nodes(&value)?;
81 cluster_required(&mut config, CLUSTER_SEED_NODES)?.seed_nodes = seed_nodes;
82 }
83 CLUSTER_LISTEN_ADDRESS => {
84 let listen_address = parse_socket_addr(CLUSTER_LISTEN_ADDRESS, &value)?;
85 cluster_required(&mut config, CLUSTER_LISTEN_ADDRESS)?.listen_address =
86 listen_address;
87 }
88 CLUSTER_COOKIE => {
89 let cookie = env_string(CLUSTER_COOKIE, &value)?;
90 cluster_required(&mut config, CLUSTER_COOKIE)?.cookie = cookie;
91 }
92 AUTH_TOKEN => {
93 let token = env_string(AUTH_TOKEN, &value)?;
98 config.auth = Some(super::types::AuthConfig { token });
99 }
100 _ => {}
101 }
102 }
103
104 Ok(config)
105}
106
107fn parse_socket_addr(name: &str, value: &OsStr) -> Result<SocketAddr, ServerError> {
108 let value = env_string(name, value)?;
109 value.parse::<SocketAddr>().map_err(|error| {
110 config_load(format!(
111 "environment variable {name} must be a socket address: {error}"
112 ))
113 })
114}
115
116fn parse_u64(name: &str, value: &OsStr) -> Result<u64, ServerError> {
117 let value = env_string(name, value)?;
118 value.parse::<u64>().map_err(|error| {
119 config_load(format!(
120 "environment variable {name} must be an unsigned integer: {error}"
121 ))
122 })
123}
124
125fn parse_seed_nodes(value: &OsStr) -> Result<Vec<SocketAddr>, ServerError> {
126 let value = env_string(CLUSTER_SEED_NODES, value)?;
127 if value.trim().is_empty() {
128 return Ok(Vec::new());
129 }
130
131 value
132 .split(',')
133 .enumerate()
134 .map(|(index, candidate)| parse_seed_node(index, candidate))
135 .collect()
136}
137
138fn parse_seed_node(index: usize, candidate: &str) -> Result<SocketAddr, ServerError> {
139 let candidate = candidate.trim();
140 if candidate.is_empty() {
141 return Err(config_load(format!(
142 "environment variable {CLUSTER_SEED_NODES} contains an empty seed node at position {}",
143 index + 1
144 )));
145 }
146
147 candidate.parse::<SocketAddr>().map_err(|error| {
148 config_load(format!(
149 "environment variable {CLUSTER_SEED_NODES} contains invalid seed node '{}' at position {}: {error}",
150 candidate,
151 index + 1
152 ))
153 })
154}
155
156fn env_string(name: &str, value: &OsStr) -> Result<String, ServerError> {
157 value.to_str().map(str::to_owned).ok_or_else(|| {
158 config_load(format!(
159 "environment variable {name} contains non-Unicode data"
160 ))
161 })
162}
163
164fn cluster_required<'a>(
165 config: &'a mut ServerConfig,
166 name: &str,
167) -> Result<&'a mut super::types::ClusterConfig, ServerError> {
168 config
169 .cluster
170 .as_mut()
171 .ok_or_else(|| ServerError::ConfigValidation {
172 message: format!(
173 "environment variable {name} requires a [cluster] section in the configuration file"
174 ),
175 })
176}
177
178const fn config_load(message: String) -> ServerError {
179 ServerError::ConfigLoad { message }
180}
181
182#[cfg(test)]
183mod tests {
184 use std::ffi::OsString;
185 use std::net::SocketAddr;
186 use std::path::{Path, PathBuf};
187
188 use crate::ServerError;
189
190 use super::apply_env_overrides_from;
191 use crate::config::types::{ChannelDef, ClusterConfig, RoutingRuleDef, ServerConfig};
192 use crate::config::{load_from_file, validate};
193
194 fn socket(address: &str) -> Result<SocketAddr, Box<dyn std::error::Error>> {
195 Ok(address.parse()?)
196 }
197
198 fn sample_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
199 Ok(ServerConfig {
200 listen_address: socket("127.0.0.1:8080")?,
201 health_listen_address: socket("127.0.0.1:8081")?,
202 drain_timeout_ms: 30_000,
203 channels: vec![ChannelDef {
204 name: "orders".to_owned(),
205 schema_ref: None,
206 durable: true,
207 loaded_schema: None,
208 }],
209 routing_rules: vec![RoutingRuleDef {
210 source_channel: "orders".to_owned(),
211 target_channel: "orders".to_owned(),
212 predicate: None,
213 }],
214 persistence_path: Some(PathBuf::from("/tmp")),
215 cluster: Some(ClusterConfig {
216 node_name: "node-a".to_owned(),
217 listen_address: socket("127.0.0.1:9000")?,
218 seed_nodes: vec![socket("127.0.0.1:9001")?],
219 cookie: "test-cookie".to_owned(),
220 }),
221 auth: None,
222 services: crate::config::types::ServicesConfig::default(),
223 limits: crate::config::types::LimitsConfig::default(),
224 })
225 }
226
227 fn env_pair(name: &str, value: &str) -> (OsString, OsString) {
228 (OsString::from(name), OsString::from(value))
229 }
230
231 fn write_temp_config(contents: &str) -> Result<PathBuf, Box<dyn std::error::Error>> {
232 let path = std::env::temp_dir().join(format!(
233 "liminal-server-env-pipeline-{}.toml",
234 std::process::id()
235 ));
236 std::fs::write(&path, contents)?;
237 Ok(path)
238 }
239
240 fn remove_temp_file(path: &Path) -> Result<(), Box<dyn std::error::Error>> {
241 if path.exists() {
242 std::fs::remove_file(path)?;
243 }
244 Ok(())
245 }
246
247 #[test]
248 fn listen_address_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
249 let config = sample_config()?;
250 let config = apply_env_overrides_from(
251 config,
252 vec![env_pair("LIMINAL_LISTEN_ADDRESS", "0.0.0.0:9090")],
253 )?;
254
255 assert_eq!(config.listen_address, socket("0.0.0.0:9090")?);
256
257 Ok(())
258 }
259
260 #[test]
261 fn health_listen_address_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>>
262 {
263 let config = sample_config()?;
264 let config = apply_env_overrides_from(
265 config,
266 vec![env_pair("LIMINAL_HEALTH_LISTEN_ADDRESS", "0.0.0.0:9191")],
267 )?;
268
269 assert_eq!(config.health_listen_address, socket("0.0.0.0:9191")?);
270
271 Ok(())
272 }
273
274 #[test]
275 fn drain_timeout_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
276 let config = sample_config()?;
277 let config =
278 apply_env_overrides_from(config, vec![env_pair("LIMINAL_DRAIN_TIMEOUT_MS", "1250")])?;
279
280 assert_eq!(config.drain_timeout_ms, 1250);
281
282 Ok(())
283 }
284
285 #[test]
286 fn persistence_path_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
287 let config = sample_config()?;
288 let config = apply_env_overrides_from(
289 config,
290 vec![env_pair("LIMINAL_PERSISTENCE_PATH", "/var/lib/liminal")],
291 )?;
292
293 assert_eq!(
294 config.persistence_path.as_deref(),
295 Some(Path::new("/var/lib/liminal"))
296 );
297
298 Ok(())
299 }
300
301 #[test]
302 fn cluster_overrides_replace_existing_cluster_values() -> Result<(), Box<dyn std::error::Error>>
303 {
304 let config = sample_config()?;
305 let config = apply_env_overrides_from(
306 config,
307 vec![
308 env_pair("LIMINAL_CLUSTER_NODE_NAME", "node-b"),
309 env_pair(
310 "LIMINAL_CLUSTER_SEED_NODES",
311 "127.0.0.1:9100, 127.0.0.1:9200",
312 ),
313 ],
314 )?;
315
316 let Some(cluster) = config.cluster else {
317 return Err("cluster config should remain present".into());
318 };
319 assert_eq!(cluster.node_name, "node-b");
320 assert_eq!(cluster.seed_nodes.len(), 2);
321 assert_eq!(cluster.seed_nodes[0], socket("127.0.0.1:9100")?);
322 assert_eq!(cluster.seed_nodes[1], socket("127.0.0.1:9200")?);
323
324 Ok(())
325 }
326
327 #[test]
328 fn cluster_listen_address_and_cookie_overrides_replace_values()
329 -> Result<(), Box<dyn std::error::Error>> {
330 let config = sample_config()?;
331 let config = apply_env_overrides_from(
332 config,
333 vec![
334 env_pair("LIMINAL_CLUSTER_LISTEN_ADDRESS", "127.0.0.1:9500"),
335 env_pair("LIMINAL_CLUSTER_COOKIE", "override-cookie"),
336 ],
337 )?;
338
339 let Some(cluster) = config.cluster else {
340 return Err("cluster config should remain present".into());
341 };
342 assert_eq!(cluster.listen_address, socket("127.0.0.1:9500")?);
343 assert_eq!(cluster.cookie, "override-cookie");
344
345 Ok(())
346 }
347
348 #[test]
349 fn cluster_listen_address_override_without_cluster_section_returns_validation_error()
350 -> Result<(), Box<dyn std::error::Error>> {
351 let mut config = sample_config()?;
352 config.cluster = None;
353 let result = apply_env_overrides_from(
354 config,
355 vec![env_pair("LIMINAL_CLUSTER_LISTEN_ADDRESS", "127.0.0.1:9500")],
356 );
357
358 assert!(matches!(result, Err(ServerError::ConfigValidation { .. })));
359
360 Ok(())
361 }
362
363 #[test]
364 fn auth_token_override_replaces_existing_token() -> Result<(), Box<dyn std::error::Error>> {
365 let mut config = sample_config()?;
366 config.auth = Some(crate::config::types::AuthConfig {
367 token: "file-token".to_owned(),
368 });
369
370 let config =
371 apply_env_overrides_from(config, vec![env_pair("LIMINAL_AUTH_TOKEN", "env-token")])?;
372
373 let auth = config.auth.ok_or("auth section should remain present")?;
374 assert_eq!(auth.token, "env-token");
375
376 Ok(())
377 }
378
379 #[test]
380 fn auth_token_override_creates_missing_auth_section() -> Result<(), Box<dyn std::error::Error>>
381 {
382 let mut config = sample_config()?;
383 config.auth = None;
384
385 let config =
386 apply_env_overrides_from(config, vec![env_pair("LIMINAL_AUTH_TOKEN", "env-token")])?;
387
388 let auth = config.auth.ok_or("auth section should have been created")?;
390 assert_eq!(auth.token, "env-token");
391
392 Ok(())
393 }
394
395 #[test]
396 fn absent_environment_variables_leave_config_unchanged()
397 -> Result<(), Box<dyn std::error::Error>> {
398 let config = sample_config()?;
399 let original_address = config.listen_address;
400 let original_health_address = config.health_listen_address;
401 let original_drain_timeout_ms = config.drain_timeout_ms;
402 let original_path = config.persistence_path.clone();
403 let original_cluster_name = config
404 .cluster
405 .as_ref()
406 .map(|cluster| cluster.node_name.clone());
407
408 let config = apply_env_overrides_from(config, Vec::new())?;
409
410 assert_eq!(config.listen_address, original_address);
411 assert_eq!(config.health_listen_address, original_health_address);
412 assert_eq!(config.drain_timeout_ms, original_drain_timeout_ms);
413 assert_eq!(config.persistence_path, original_path);
414 assert_eq!(
415 config
416 .cluster
417 .as_ref()
418 .map(|cluster| cluster.node_name.clone()),
419 original_cluster_name
420 );
421
422 Ok(())
423 }
424
425 #[test]
426 fn invalid_listen_address_override_returns_config_load()
427 -> Result<(), Box<dyn std::error::Error>> {
428 let config = sample_config()?;
429 let result = apply_env_overrides_from(
430 config,
431 vec![env_pair("LIMINAL_LISTEN_ADDRESS", "not-a-socket")],
432 );
433
434 assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
435
436 Ok(())
437 }
438
439 #[test]
440 fn invalid_health_listen_address_override_returns_config_load()
441 -> Result<(), Box<dyn std::error::Error>> {
442 let config = sample_config()?;
443 let result = apply_env_overrides_from(
444 config,
445 vec![env_pair("LIMINAL_HEALTH_LISTEN_ADDRESS", "not-a-socket")],
446 );
447
448 assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
449
450 Ok(())
451 }
452
453 #[test]
454 fn invalid_drain_timeout_override_returns_config_load() -> Result<(), Box<dyn std::error::Error>>
455 {
456 let config = sample_config()?;
457 let result = apply_env_overrides_from(
458 config,
459 vec![env_pair("LIMINAL_DRAIN_TIMEOUT_MS", "not-a-number")],
460 );
461
462 assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
463
464 Ok(())
465 }
466
467 #[test]
468 fn cluster_override_without_cluster_section_returns_validation_error()
469 -> Result<(), Box<dyn std::error::Error>> {
470 let mut config = sample_config()?;
471 config.cluster = None;
472 let result = apply_env_overrides_from(
473 config,
474 vec![env_pair("LIMINAL_CLUSTER_NODE_NAME", "node-b")],
475 );
476
477 assert!(matches!(result, Err(ServerError::ConfigValidation { .. })));
478
479 Ok(())
480 }
481
482 #[test]
483 fn file_then_env_then_validate_pipeline_gives_env_precedence()
484 -> Result<(), Box<dyn std::error::Error>> {
485 let toml = r#"
486listen_address = "127.0.0.1:8080"
487health_listen_address = "127.0.0.1:8081"
488drain_timeout_ms = 30000
489persistence_path = "/tmp"
490
491[[channels]]
492name = "orders"
493durable = true
494
495[[routing_rules]]
496source_channel = "orders"
497target_channel = "orders"
498"#;
499 let path = write_temp_config(toml)?;
500 let config = load_from_file(&path)?;
501 let mut config = apply_env_overrides_from(
502 config,
503 vec![env_pair("LIMINAL_LISTEN_ADDRESS", "0.0.0.0:9090")],
504 )?;
505 validate(&mut config, path.parent())?;
506 remove_temp_file(&path)?;
507
508 assert_eq!(config.listen_address, socket("0.0.0.0:9090")?);
509
510 Ok(())
511 }
512}