aster_forge_config/notification/
config.rs

1use serde::{Deserialize, Serialize};
2#[cfg(feature = "redis-pubsub")]
3use std::sync::Arc;
4
5use crate::{ConfigCoreError, Result};
6
7#[cfg(feature = "redis-pubsub")]
8use super::notifier::RedisConfigChangeNotifier;
9#[cfg(feature = "redis-pubsub")]
10use super::notifier::SharedConfigChangeNotifier;
11use super::runtime::ConfigSyncRuntime;
12
13/// Disabled config-sync backend name.
14pub const CONFIG_SYNC_BACKEND_DISABLED: &str = "disabled";
15/// Redis pub/sub config-sync backend name.
16pub const CONFIG_SYNC_BACKEND_REDIS: &str = "redis";
17
18/// Config-sync broker endpoint input.
19///
20/// Existing string endpoints remain valid. The structured form carries a base URL without
21/// userinfo and raw credentials that the selected transport injects safely.
22#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
23#[serde(untagged)]
24pub enum ConfigSyncEndpoint {
25    /// A complete broker URL.
26    Url(String),
27    /// A base broker URL without userinfo plus raw credentials.
28    Credentials {
29        /// Absolute broker URL without username or password.
30        base_url: String,
31        /// Raw broker username.
32        #[serde(default, skip_serializing)]
33        username: Option<String>,
34        /// Raw broker password.
35        #[serde(default, skip_serializing)]
36        password: Option<String>,
37    },
38}
39
40impl ConfigSyncEndpoint {
41    /// Creates a complete-URL endpoint.
42    pub fn url(url: impl Into<String>) -> Self {
43        Self::Url(url.into())
44    }
45
46    /// Creates a base URL plus raw credentials endpoint.
47    pub fn credentials(
48        base_url: impl Into<String>,
49        username: Option<String>,
50        password: Option<String>,
51    ) -> Self {
52        Self::Credentials {
53            base_url: base_url.into(),
54            username,
55            password,
56        }
57    }
58}
59
60impl Default for ConfigSyncEndpoint {
61    fn default() -> Self {
62        Self::Url(String::new())
63    }
64}
65
66impl From<String> for ConfigSyncEndpoint {
67    fn from(url: String) -> Self {
68        Self::Url(url)
69    }
70}
71
72impl From<&str> for ConfigSyncEndpoint {
73    fn from(url: &str) -> Self {
74        Self::Url(url.to_string())
75    }
76}
77
78impl std::fmt::Debug for ConfigSyncEndpoint {
79    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        match self {
81            Self::Url(_) => formatter.write_str("ConfigSyncEndpoint::Url(<redacted>)"),
82            Self::Credentials { .. } => {
83                formatter.write_str("ConfigSyncEndpoint::Credentials(<redacted>)")
84            }
85        }
86    }
87}
88
89/// Static configuration for cross-process config reload synchronization.
90///
91/// The field names describe a generic broker contract instead of a Redis-only
92/// shape. Current services can map `backend = "redis"` to Redis pub/sub, while
93/// future `RabbitMQ`, NATS, or other transports can reuse the same product config
94/// surface and add backend-specific interpretation behind the notifier factory.
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96pub struct ConfigSyncConfig {
97    /// Transport backend name, for example `disabled` or `redis`.
98    #[serde(default = "ConfigSyncConfig::default_backend")]
99    pub backend: String,
100    /// Broker endpoint URL. Redis uses a Redis URL.
101    #[serde(default)]
102    pub endpoint: ConfigSyncEndpoint,
103    /// Logical reload topic. Transports may map this to a channel, exchange,
104    /// subject, or routing key.
105    #[serde(default = "ConfigSyncConfig::default_topic")]
106    pub topic: String,
107}
108
109impl Default for ConfigSyncConfig {
110    fn default() -> Self {
111        Self {
112            backend: Self::default_backend(),
113            endpoint: ConfigSyncEndpoint::default(),
114            topic: Self::default_topic(),
115        }
116    }
117}
118
119impl ConfigSyncConfig {
120    /// Returns the default disabled backend name.
121    #[must_use]
122    pub fn default_backend() -> String {
123        CONFIG_SYNC_BACKEND_DISABLED.to_string()
124    }
125
126    /// Returns the default logical reload topic.
127    #[must_use]
128    pub fn default_topic() -> String {
129        "aster.config_reload".to_string()
130    }
131
132    /// Returns whether cross-process sync is enabled.
133    #[must_use]
134    pub fn enabled(&self) -> bool {
135        !matches!(
136            self.backend.trim().to_ascii_lowercase().as_str(),
137            "" | "disabled" | "none"
138        )
139    }
140}
141
142/// Returns the conventional config-sync topic for a product namespace.
143#[must_use]
144pub fn default_config_sync_topic(namespace: &str) -> String {
145    format!("{}.config_reload", namespace.trim())
146}
147
148/// Builds a namespaced config-sync runtime from static config.
149///
150/// This common backend factory owns backend dispatch, runtime ID generation, and
151/// transport-specific topic mapping. Product crates only pass their namespace
152/// and provide their reload callback to [`ConfigSyncRuntime::run_reload_subscription`].
153///
154/// # Errors
155///
156/// Returns [`ConfigError`] when the sync backend, endpoint, topic, credentials, or runtime id is invalid.
157pub fn build_config_sync_runtime(
158    config: &ConfigSyncConfig,
159    namespace: &str,
160) -> Result<ConfigSyncRuntime> {
161    build_config_sync_runtime_with_runtime_id(
162        config,
163        namespace,
164        aster_forge_utils::id::new_runtime_id(),
165    )
166}
167
168/// Builds a namespaced config-sync runtime with an explicit runtime ID.
169///
170/// Products normally use [`build_config_sync_runtime`]. This variant is useful when the product
171/// already has a stable process identity or when tests need deterministic self-origin filtering.
172///
173/// # Errors
174///
175/// Returns [`ConfigError`] when the sync backend, endpoint, topic, credentials, or runtime id is invalid.
176pub fn build_config_sync_runtime_with_runtime_id(
177    config: &ConfigSyncConfig,
178    namespace: &str,
179    runtime_id: impl Into<String>,
180) -> Result<ConfigSyncRuntime> {
181    let namespace = namespace.trim();
182    let runtime_id = runtime_id.into();
183    let topic = config_sync_topic(config, namespace);
184    match config.backend.trim().to_ascii_lowercase().as_str() {
185        "" | "disabled" | "none" => Ok(ConfigSyncRuntime::disabled_with_runtime_id(
186            namespace, runtime_id,
187        )),
188        CONFIG_SYNC_BACKEND_REDIS => {
189            build_redis_config_sync_runtime(config, namespace, runtime_id, &topic)
190        }
191        backend => Err(ConfigCoreError::invalid_value(format!(
192            "unsupported config_sync.backend '{backend}'"
193        ))),
194    }
195}
196fn config_sync_topic(config: &ConfigSyncConfig, namespace: &str) -> String {
197    let topic = config.topic.trim();
198    if topic.is_empty() || topic == ConfigSyncConfig::default_topic() {
199        default_config_sync_topic(namespace)
200    } else {
201        topic.to_string()
202    }
203}
204
205#[cfg(feature = "redis-pubsub")]
206fn build_redis_config_sync_runtime(
207    config: &ConfigSyncConfig,
208    namespace: &str,
209    runtime_id: String,
210    topic: &str,
211) -> Result<ConfigSyncRuntime> {
212    let channel = redis_channel_from_topic(topic);
213    let notifier = match &config.endpoint {
214        ConfigSyncEndpoint::Url(endpoint) => {
215            if endpoint.trim().is_empty() {
216                return Err(ConfigCoreError::invalid_value(
217                    "config_sync.endpoint is required when config_sync.backend is redis",
218                ));
219            }
220            RedisConfigChangeNotifier::from_url(endpoint.trim(), channel)?
221        }
222        ConfigSyncEndpoint::Credentials {
223            base_url,
224            username,
225            password,
226        } => {
227            if base_url.trim().is_empty() {
228                return Err(ConfigCoreError::invalid_value(
229                    "config_sync.endpoint base_url is required when config_sync.backend is redis",
230                ));
231            }
232            RedisConfigChangeNotifier::from_credentials(
233                base_url.trim(),
234                username.as_deref(),
235                password.as_deref(),
236                channel,
237            )?
238        }
239    };
240    Ok(ConfigSyncRuntime::new(
241        namespace,
242        runtime_id,
243        Arc::new(notifier) as SharedConfigChangeNotifier,
244    ))
245}
246
247#[cfg(not(feature = "redis-pubsub"))]
248fn build_redis_config_sync_runtime(
249    _config: &ConfigSyncConfig,
250    _namespace: &str,
251    _runtime_id: String,
252    _topic: &str,
253) -> Result<ConfigSyncRuntime> {
254    Err(ConfigCoreError::invalid_value(
255        "config_sync.backend 'redis' requires the redis-pubsub feature",
256    ))
257}
258
259#[cfg(any(feature = "redis-pubsub", test))]
260pub(super) fn redis_channel_from_topic(topic: &str) -> String {
261    topic.trim().replace('.', ":")
262}
aster_forge_config/notification/config.rs

aster_forge_config/notification/
config.rs
